提交

.gitignore

@@ -0,0 +1,47 @@
+######################################################################
+# Build Tools
+
+.gradle
+/build/
+!gradle/wrapper/gradle-wrapper.jar
+
+target/
+!.mvn/wrapper/maven-wrapper.jar
+
+######################################################################
+# IDE
+
+### STS ###
+.apt_generated
+.classpath
+.factorypath
+.project
+.settings
+.springBeans
+
+### IntelliJ IDEA ###
+.idea
+*.iws
+*.iml
+*.ipr
+
+### JRebel ###
+rebel.xml
+### NetBeans ###
+nbproject/private/
+build/*
+nbbuild/
+dist/
+nbdist/
+.nb-gradle/
+
+######################################################################
+# Others
+*.log
+*.xml.versionsBackup
+*.swp
+
+!*/build/*.java
+!*/build/*.html
+!*/build/*.xml
+/venv/

SparkApi.py

@@ -0,0 +1,136 @@
+import _thread as thread
+import base64
+import datetime
+import hashlib
+import hmac
+import json
+from urllib.parse import urlparse
+import ssl
+from datetime import datetime
+from time import mktime
+from urllib.parse import urlencode
+from wsgiref.handlers import format_date_time
+
+import websocket  # 使用websocket_client
+answer = ""
+
+class Ws_Param(object):
+    # 初始化
+    def __init__(self, APPID, APIKey, APISecret, Spark_url):
+        self.APPID = APPID
+        self.APIKey = APIKey
+        self.APISecret = APISecret
+        self.host = urlparse(Spark_url).netloc
+        self.path = urlparse(Spark_url).path
+        self.Spark_url = Spark_url
+
+    # 生成url
+    def create_url(self):
+        # 生成RFC1123格式的时间戳
+        now = datetime.now()
+        date = format_date_time(mktime(now.timetuple()))
+
+        # 拼接字符串
+        signature_origin = "host: " + self.host + "\n"
+        signature_origin += "date: " + date + "\n"
+        signature_origin += "GET " + self.path + " HTTP/1.1"
+
+        # 进行hmac-sha256进行加密
+        signature_sha = hmac.new(self.APISecret.encode('utf-8'), signature_origin.encode('utf-8'),
+                                 digestmod=hashlib.sha256).digest()
+
+        signature_sha_base64 = base64.b64encode(signature_sha).decode(encoding='utf-8')
+
+        authorization_origin = f'api_key="{self.APIKey}", algorithm="hmac-sha256", headers="host date request-line", signature="{signature_sha_base64}"'
+
+        authorization = base64.b64encode(authorization_origin.encode('utf-8')).decode(encoding='utf-8')
+
+        # 将请求的鉴权参数组合为字典
+        v = {
+            "authorization": authorization,
+            "date": date,
+            "host": self.host
+        }
+        # 拼接鉴权参数，生成url
+        url = self.Spark_url + '?' + urlencode(v)
+        # 此处打印出建立连接时候的url,参考本demo的时候可取消上方打印的注释，比对相同参数时生成的url与自己代码生成的url是否一致
+        return url
+
+
+# 收到websocket错误的处理
+def on_error(ws, error):
+    print("### error:", error)
+
+
+# 收到websocket关闭的处理
+def on_close(ws,one,two):
+    print(" ")
+
+
+# 收到websocket连接建立的处理
+def on_open(ws):
+    thread.start_new_thread(run, (ws,))
+
+
+def run(ws, *args):
+    data = json.dumps(gen_params(appid=ws.appid, domain= ws.domain,question=ws.question))
+    ws.send(data)
+
+
+# 收到websocket消息的处理
+def on_message(ws, message):
+    # print(message)
+    data = json.loads(message)
+    code = data['header']['code']
+    if code != 0:
+        print(f'请求错误: {code}, {data}')
+        ws.close()
+    else:
+        choices = data["payload"]["choices"]
+        status = choices["status"]
+        content = choices["text"][0]["content"]
+        print(content,end ="")
+        global answer
+        answer += content
+        # print(1)
+        if status == 2:
+            ws.close()
+
+
+def gen_params(appid, domain,question):
+    """
+    通过appid和用户的提问来生成请参数
+    """
+    data = {
+        "header": {
+            "app_id": appid,
+            "uid": "1234"
+        },
+        "parameter": {
+            "chat": {
+                "domain": domain,
+                "temperature": 0.5,
+                "max_tokens": 2048
+            }
+        },
+        "payload": {
+            "message": {
+                "text": question
+            }
+        }
+    }
+    return data
+
+
+def main(appid, api_key, api_secret, Spark_url,domain, question):
+    # print("星火:")
+    wsParam = Ws_Param(appid, api_key, api_secret, Spark_url)
+    websocket.enableTrace(False)
+    wsUrl = wsParam.create_url()
+    ws = websocket.WebSocketApp(wsUrl, on_message=on_message, on_error=on_error, on_close=on_close, on_open=on_open)
+    ws.appid = appid
+    ws.question = question
+    ws.domain = domain
+    ws.run_forever(sslopt={"cert_reqs": ssl.CERT_NONE})
+
+

main.py

@@ -0,0 +1,71 @@
+from fastapi import FastAPI
+import requests
+from flask import request
+from requests_toolbelt import MultipartEncoder
+
+import renamePhotoBatch
+
+app = FastAPI()
+
+
+@app.get("/")
+async def root():
+    return {"message": "Hello World"}
+
+
+@app.get("/hello/{name}")
+async def say_hello(name: str):
+    return {"message": f"Hello {name}"}
+
+@app.post("/rename_photo")
+async def rename_photo():
+    return renamePhotoBatch.solution(request.form.get('path'))
+
+# 获取文章列表
+def get_post_list_urls(page, username):
+    headers = {
+        "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:62.0) Gecko/20100101 Firefox/62.0"
+    }
+    lb_params = {
+        "page": page,
+        "size": "20",
+        "businessType": "blog",
+        "orderby": "",
+        "noMore": "false",
+        "year": "",
+        "month": "",
+        "username": username,
+    }
+    urls = "https://blog.csdn.net/community/home-api/v1/get-business-list"
+    try:
+        res = requests.get(url=urls, headers=headers, params=lb_params)
+        # print(res.status_code)
+        data_list = res.json()["data"]["list"]
+        return data_list
+    except Exception as e:
+        print(f"[+] [ending] [{e}]")
+        return 0
+
+# 查询质量分数
+def get_sorce(article_url):
+    url = "https://bizapi.csdn.net/trends/api/v1/get-article-score"
+
+    headers = {
+    	# headers
+    }
+    params = MultipartEncoder({"url": article_url})
+    headers["Content-Type"] = params.content_type
+    try:
+        res = requests.post(
+            url,
+            headers=headers,
+            data=params,
+            timeout=10,
+        )
+        # print(f"[+] [{article_url}] [{res.status_code}] ")
+        sorce = res.json()["data"]["score"]
+        # print(sorce)
+        return sorce
+    except Exception as e:
+        print(f"[+] [分数获取失败] [{article_url}] [{res.status_code}] [{e}]")
+        return 0

renamePhotoBatch.py

@@ -0,0 +1,45 @@
+#!/usr/bin/env python3
+# @author 轩辕龙儿
+# @date 2023/10/23
+# @file renamePhotoBatch.py
+import os
+import glob
+import exifread
+
+
+# 获取照片的拍摄时间
+def get_photo_taken_time(image_path):
+    with open(image_path, 'rb') as f:
+        tags = exifread.process_file(f)
+        if 'EXIF DateTimeOriginal' in tags:
+            taken_time = tags['EXIF DateTimeOriginal']
+            return taken_time
+        else:
+            return None
+
+
+def solution():
+    # 获取当前时间并格式化为年月日时分秒毫秒
+
+    # 设置照片所在的文件夹路径
+    folder_path = "C:\\Users\\hyy\\Pictures\\图片"
+
+    # 获取文件夹中的所有照片文件
+    photo_files = glob.glob(os.path.join(folder_path, "*"))
+
+    # 遍历照片文件并重命名
+    for photo in photo_files:
+        creation_time_stamp = get_photo_taken_time(photo)
+        if creation_time_stamp is None:
+            continue
+        formatted_time = "IMG_" + creation_time_stamp.values.replace(':', '').replace(' ', '_')
+        file_name, file_ext = os.path.splitext(photo)
+        new_file_name = f"{formatted_time}{file_ext}"
+        new_file_path = os.path.join(folder_path, new_file_name)
+        os.rename(photo, new_file_path)
+
+    print("照片重命名完成！")
+
+
+if __name__ == "__main__":
+    solution()