From 0426a77764ad5f33d0456d2d6dbe927bec81e9f9 Mon Sep 17 00:00:00 2001 From: liubiren Date: Tue, 6 Jan 2026 17:54:56 +0800 Subject: [PATCH] 260106 from nuc --- utils/__init__.py | 2 +- utils/request.py | 36 ++++++++--------- 短视频合成自动化/draft.py | 75 +++++++++++++++++------------------- 短视频合成自动化/edgetts.py | 16 ++++---- 短视频合成自动化/export.py | 51 +++++++++++++++--------- 票据理赔自动化/image.py | 6 +-- 票据理赔自动化/masterdata.py | 2 + 7 files changed, 98 insertions(+), 90 deletions(-) diff --git a/utils/__init__.py b/utils/__init__.py index d91180a..aad0a9d 100644 --- a/utils/__init__.py +++ b/utils/__init__.py @@ -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 \ No newline at end of file diff --git a/utils/request.py b/utils/request.py index 4dd9f1e..96ab79e 100644 --- a/utils/request.py +++ b/utils/request.py @@ -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: 使用流式传输 diff --git a/短视频合成自动化/draft.py b/短视频合成自动化/draft.py index 4e3fb19..3a258d7 100644 --- a/短视频合成自动化/draft.py +++ b/短视频合成自动化/draft.py @@ -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 diff --git a/短视频合成自动化/edgetts.py b/短视频合成自动化/edgetts.py index c0ed9b9..eb95de6 100644 --- a/短视频合成自动化/edgetts.py +++ b/短视频合成自动化/edgetts.py @@ -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 diff --git a/短视频合成自动化/export.py b/短视频合成自动化/export.py index e07990f..5c94b34 100644 --- a/短视频合成自动化/export.py +++ b/短视频合成自动化/export.py @@ -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 diff --git a/票据理赔自动化/image.py b/票据理赔自动化/image.py index fc7766e..e4563a1 100644 --- a/票据理赔自动化/image.py +++ b/票据理赔自动化/image.py @@ -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 # 实例化认证器 diff --git a/票据理赔自动化/masterdata.py b/票据理赔自动化/masterdata.py index 6311068..0fc557f 100644 --- a/票据理赔自动化/masterdata.py +++ b/票据理赔自动化/masterdata.py @@ -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