liubiren
/
Python
1
1
Fork 0
Code Issues Pull Requests Packages Projects Releases Activity

260106 from nuc

This commit is contained in:
liubiren 2026-01-06 17:54:56 +08:00
parent 7379c992bf
commit 0426a77764
7 changed files with 98 additions and 90 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
utils/__init__.py
View File

@ -2,4 +2,4 @@ from mysql import MySQL
from sqlite import SQLite
from request import restrict, Authenticator, Request
from feishu import Feishu
from rules_engine import RulesEngine
from rules_engine import RulesEngine

36
utils/request.py
View File

@ -17,9 +17,15 @@ from requests import Response, Session
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry
from sqlite import SQLite
# 基于令牌桶限流算法的装饰器
def restrict(refill_rate: float = 5.0, max_tokens: int = 5):
"""
请求限速装饰器
:param refill_rate: 令牌填充速率，单位为个/秒
:param max_tokens: 最大令牌数，单位为个
"""
class TokenBucket:
@ -178,15 +184,15 @@ class Request:
raise ValueError("上传文件和使用流式传输不能同时使用")
return self
class CacheClient(SQLiteClient):
"""缓存客户端"""
class Caches(SQLite):
"""请求缓存"""
def __init__(self, cache_ttl: int):
"""
初始化缓存数据库
初始化
:param cache_ttl: 缓存生存时间，单位为秒
"""
# 初始化SQLite客户端
# 初始化
super().__init__(database=Path(__file__).parent.resolve() / "caches.db")
# 初始化缓存生存时间，单位为秒
self.cache_ttl = cache_ttl
@ -293,11 +299,10 @@ class Request:
# 初始化缓存生存时间，单位由天转为秒
self.cache_ttl = cache_ttl * 86400
self.cache_client: Optional[HTTPClient.CacheClient] = None
# 若使用缓存则实例化缓存客户端
self.caches: Optional[Request.Caches] = None
# 若使用缓存则实例化缓存
if self.cache_enabled:
# 初始化缓存客户端
self.cache_client = self.CacheClient(cache_ttl=self.cache_ttl)
self.caches = Request.Caches(cache_ttl=self.cache_ttl)
def __del__(self):
"""析构时关闭请求会话"""
@ -346,26 +351,19 @@ class Request:
def get(
self, **kwargs
) -> Union[str, Tuple[str, bytes], Dict[str, Any], ElementTree.Element, None]:
) -> Any:
"""发送GET请求"""
return self._request(method="GET", parameters=self.Parameters(**kwargs))
def post(
self, **kwargs
) -> Union[str, Tuple[str, bytes], Dict[str, Any], ElementTree.Element, None]:
) -> Any:
"""发送POST请求"""
return self._request(method="POST", parameters=self.Parameters(**kwargs))
def download(
self, stream_enabled: bool = False, chunk_size: int = 1024, **kwargs
) -> Union[
str,
Tuple[str, bytes],
Dict[str, Any],
ElementTree.Element,
Generator[bytes, None, None],
None,
]:
) -> Any:
"""
下载文件
:param stream_enabled: 使用流式传输

75
短视频合成自动化/draft.py
View File

@ -1,4 +1,7 @@
# -*- coding: utf-8 -*-
"""
draft模块
"""
from pathlib import Path
from typing import Any, Dict, List, Optional, Tuple, Union
@ -19,7 +22,8 @@ class JianYingDraft:
6、将草稿保存至剪映草稿文件夹内
"""
# noinspection PyShadowingNames
# pylint: disable=too-many-arguments
# pylint: disable=too-many-positional-arguments
def __init__(
self,
drafts_folder: pyJianYingDraft.DraftFolder,
@ -40,7 +44,6 @@ class JianYingDraft:
:param video_fps: 视频帧率（单位为帧/秒），默认为 30
:param materials_folder_path: 素材文件夹路径
"""
# noinspection PyBroadException
try:
# 新建草稿
self.draft = drafts_folder.create_draft(
@ -56,14 +59,14 @@ class JianYingDraft:
self.materials_folder_path = materials_folder_path
except Exception as exception:
raise RuntimeError(f"发生异常：{str(exception)}")
raise RuntimeError(f"发生异常：{str(exception)}") from exception
def add_text_segment(
self,
track_name: str,
text: str,
add_track: bool = True,
timerange: Optional[Tuple[Optional[int, str], Optional[int, str]]] = None,
timerange: Optional[Tuple[Union[int, str], Union[int, str]]] = None,
font: Optional[str] = None,
style: Optional[Dict[str, Any]] = None,
border: Optional[Dict[str, Any]] = None,
@ -132,20 +135,18 @@ class JianYingDraft:
# 向指定文本轨道添加文本片段
self.draft.add_segment(segment=text_segment, track_name=track_name)
except Exception:
raise
except Exception as exception:
raise RuntimeError(str(exception)) from exception
def add_audio_segment(
self,
track_name: str,
material_path: Path,
add_track: bool = True,
target_timerange: Optional[
Tuple[Optional[int, str], Optional[int, str]]
] = None,
target_timerange: Optional[Tuple[Union[int, str], Union[int, str]]] = None,
source_timerange: Optional[Tuple[str, str]] = None,
speed: Optional[float] = 1.0,
volume: Optional[float] = 1.0,
speed: float = 1.0,
volume: float = 1.0,
fade: Optional[Tuple[str, str]] = None,
) -> None:
"""
@ -193,28 +194,25 @@ class JianYingDraft:
# 向指定音频轨道添加音频片段
self.draft.add_segment(segment=audio_segment, track_name=track_name)
except Exception:
raise
except Exception as exception:
raise RuntimeError(str(exception)) from exception
# pylint: disable=too-many-locals
def add_video_segment(
self,
track_name: str,
material_path: Path,
target_timerange: Optional[
Tuple[Optional[int, str], Optional[int, str]]
] = None,
source_timerange: Optional[
Tuple[Optional[int, str], Optional[int, str]],
] = None,
speed: Optional[float] = 1.0,
volume: Optional[float] = 1.0,
target_timerange: Optional[Tuple[Union[int, str], Union[int, str]]] = None,
source_timerange: Optional[Tuple[Union[int, str], Union[int, str]]] = None,
speed: float = 1.0,
volume: float = 1.0,
clip_settings: Optional[Dict[str, Any]] = None,
keyframes: Optional[
List[Tuple[pyJianYingDraft.keyframe, Union[str, int], float]]
List[Tuple[pyJianYingDraft.KeyframeProperty, Union[str, int], float]]
] = None,
animation: Optional[Dict[str, Any]] = None,
transition: Optional[Dict[str, Any]] = None,
background_filling: Optional[Tuple[str, Any]] = None,
background_filling: Optional[Dict[str, Any]] = None,
) -> None:
"""
向指定视频轨道添加视频或图片片段
@ -238,9 +236,9 @@ class JianYingDraft:
track_name=track_name,
)
# 解析开始时间和持续时间
target_start, target_duration = (
target_timerange if target_timerange else (0, self.draft_duration)
# 获取持续时间
target_duration = pyJianYingDraft.time_util.tim(
(target_timerange if target_timerange else (0, self.draft_duration))[1]
)
# 视频素材
@ -276,9 +274,8 @@ class JianYingDraft:
)
# 添加关键帧
if keyframes:
# noinspection PyShadowingBuiltins
for property, offset, value in keyframes:
video_segment.add_keyframe(property, offset, value)
for _property, offset, value in keyframes:
video_segment.add_keyframe(_property, offset, value)
# 添加动画
if animation:
@ -290,23 +287,21 @@ class JianYingDraft:
# 添加背景填充
if background_filling:
video_segment.add_background_filling(*background_filling)
video_segment.add_background_filling(**background_filling)
# 向指定视频轨道添加视频或图片片段
self.draft.add_segment(segment=video_segment, track_name=track_name)
duration += video_material_duration
except Exception:
raise
except Exception as exception:
raise RuntimeError(str(exception)) from exception
def add_sticker(
self,
track_name: str,
resource_id: str,
target_timerange: Optional[
Tuple[Optional[int, str], Optional[int, str]]
] = None,
target_timerange: Optional[Tuple[Union[int, str], Union[int, str]]] = None,
clip_settings: Optional[Dict[str, Any]] = None,
) -> None:
"""
@ -344,13 +339,13 @@ class JianYingDraft:
# 向指定贴纸轨道添加贴纸片段
self.draft.add_segment(segment=sticker_segment, track_name=track_name)
except Exception:
raise
except Exception as exception:
raise RuntimeError(str(exception)) from exception
def add_subtitles(
self,
text: str,
timbre: Optional[str] = "女声-晓晓",
timbre: str = "女声-晓晓",
rate: str = "+25%",
volume: str = "+0%",
font: Optional[str] = None,
@ -429,5 +424,5 @@ class JianYingDraft:
try:
self.draft.save()
except Exception:
raise
except Exception as exception:
raise RuntimeError(str(exception)) from exception

16
短视频合成自动化/edgetts.py
View File

@ -1,17 +1,21 @@
# -*- coding: utf-8 -*-
"""
EdgeTTS模块
"""
import asyncio
from _md5 import md5
from pathlib import Path
from typing import Tuple, Union
from hashlib import md5
import edge_tts
from mutagen.mp3 import MP3
# pylint: disable=too-few-public-methods
class EdgeTTS:
"""
封装EdgeTTS，支持：
EdgeTTS模块，支持：
1、根据文本合成语音并将语音文件保存至指定文件夹内
"""
@ -41,10 +45,8 @@ class EdgeTTS:
:param timbre: 音色名称
:param rate: 语速
:param volume: 音量
:return 语音文件路径（path对象）和持续时长（单位为微秒）
:return: 语音文件路径（path对象）和持续时长（单位为微秒）
"""
# noinspection PyBroadException
try:
# 异步处理方法
async def _async_synthetize():
@ -69,4 +71,4 @@ class EdgeTTS:
except Exception as exception:
raise RuntimeError(
f"根据文本合成语音并将语音文件保存至指定文件夹内发生异常：{str(exception)}"
)
) from exception

51
短视频合成自动化/export.py
View File

@ -1,4 +1,7 @@
# -*- coding: utf-8 -*-
"""
export模块
"""
import random
import subprocess
@ -13,20 +16,23 @@ import win32gui
from draft import JianYingDraft
# pylint: disable=too-few-public-methods
# pylint: disable=too-many-instance-attributes
class JianYingExport:
"""
封装 pyJianYing中导出草稿的相关功能，支持：
封装 pyJianYingDraft.JianyingController库，支持：
1、初始化素材文件夹内所有素材
2、就工作流添加工作
3、基于工作流生成草稿
2、初始化工作流和工作配置
3、导出草稿
"""
# noinspection PyShadowingNames
# pylint: disable=too-many-arguments
# pylint: disable=too-many-positional-arguments
def __init__(
self,
materials_folder_path: str,
program_path: str = r"E:\JianyingPro\5.9.0.11632\JianyingPro.exe",
drafts_folder_path: str = r"E:\JianyingPro Drafts",
program_path: str = "E:\\JianYingPro\\5.9.0.11632\\JianYingPro.exe", # 仅可在windows运行该脚本
drafts_folder_path: str = "E:\\JianYingPro Drafts",
draft_counts: int = 10,
video_width: int = 1080,
video_height: int = 1920,
@ -34,7 +40,7 @@ class JianYingExport:
):
"""
初始化
:param program_path: 剪映执行程序路径
:param program_path: 剪映程序路径
:param drafts_folder_path: 剪映草稿文件夹路径
:param materials_folder_path: 素材文件夹路径
:param draft_counts: 草稿数，默认为 10
@ -42,11 +48,10 @@ class JianYingExport:
:param video_height: 视频高度，默认为 1920像素
:param video_fps: 视频帧率（单位为帧/秒），默认为 30
"""
# noinspection PyBroadException
try:
self.program_path = Path(program_path)
if not self.program_path.exists():
raise RuntimeError("剪映执行程序路径不存在")
raise RuntimeError("剪映程序路径不存在")
# 初始化剪映专业版进程
self.jianying_process = None
@ -309,7 +314,7 @@ class JianYingExport:
self.draft_names = []
except Exception as exception:
raise RuntimeError(f"发生异常：{str(exception)}")
raise RuntimeError(f"发生异常：{str(exception)}") from exception
def _init_materials(self) -> None:
"""
@ -500,7 +505,7 @@ class JianYingExport:
start_time = time.time()
while time.time() - start_time < timeout:
# 定位剪映执行程序窗口
# 定位剪映程序窗口
if self._locate_window() is not None:
print(f"已启动剪映专业版进程，PID {self.jianying_process.pid}")
return
@ -508,7 +513,9 @@ class JianYingExport:
raise RuntimeError("启动超时")
except Exception as exception:
raise RuntimeError(f"启动剪映专业版进程发生异常：{str(exception)}")
raise RuntimeError(
f"启动剪映专业版进程发生异常：{str(exception)}"
) from exception
def _close_process(self, timeout: int = 60) -> None:
"""
@ -517,29 +524,32 @@ class JianYingExport:
:return: 无
"""
try:
# 定位剪映执行程序窗口
# 定位剪映程序窗口
window_handle = self._locate_window()
if window_handle is not None:
# 请求关闭剪映执行程序窗口
# 请求关闭剪映程序窗口
# pylint: disable=c-extension-no-member
win32gui.SendMessage(window_handle, win32con.WM_CLOSE, 0, 0)
start_time = time.time()
while time.time() - start_time < timeout:
# pylint: disable=c-extension-no-member
if not win32gui.IsWindow(window_handle):
print("已关闭剪映专业版进程")
return
else:
time.sleep(2)
time.sleep(2)
raise RuntimeError("关闭超时")
except Exception as exception:
raise RuntimeError(f"关闭剪映专业版进程发生异常：{str(exception)}")
raise RuntimeError(
f"关闭剪映专业版进程发生异常：{str(exception)}"
) from exception
@staticmethod
def _locate_window() -> Optional[int]:
"""
定位剪映执行程序窗口
:return: 剪映执行程序窗口句柄
定位剪映程序窗口
:return: 剪映程序窗口句柄
"""
window_handle = None
@ -550,13 +560,16 @@ class JianYingExport:
# 初始化窗口句柄
nonlocal window_handle
# 获取窗口标题
# pylint: disable=c-extension-no-member
window_text = win32gui.GetWindowText(handle)
# 检查窗口是否可见且窗口标题为剪映专业版
# pylint: disable=c-extension-no-member
if win32gui.IsWindowVisible(handle) and window_text == "剪映专业版":
window_handle = handle
return False
return True
# 遍历所有顶层窗口
# pylint: disable=c-extension-no-member
win32gui.EnumWindows(callback, None)
return window_handle

6
票据理赔自动化/image.py
View File

@ -5,10 +5,10 @@ import re
from base64 import b64encode
from datetime import datetime
from decimal import Decimal, ROUND_HALF_UP
from hashlib import md5
from pathlib import Path
from typing import Optional, Tuple, Dict, Any
from hashlib import md5
import cv2
import numpy
import pandas
@ -17,8 +17,6 @@ from jionlp import parse_location
from common import dossier, master_data, rule_engine
print(1)
exit()
from utils import Authenticator, Request
# 实例化认证器

2
票据理赔自动化/masterdata.py
View File

@ -4,6 +4,8 @@ from datetime import datetime
from decimal import Decimal, ROUND_HALF_UP
from typing import Any, Dict, List, Optional
import sys
sys.path.append(".")
from utils import SQLite