huanghai/aiData
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

'commit'

Browse Source
This commit is contained in:
HuangHai
2026-01-14 07:31:31 +08:00
parent 8ab8ebb421
commit ab8ec615f2
18 changed files with 144 additions and 128 deletions
Download Patch File Download Diff File Expand all files Collapse all files

58
Apps/AiTeJiYiChong/Kit.py
View File

@@ -162,39 +162,43 @@ def click_image_template(d, template_path, timeout=5.0, threshold=0.8):
def setup_logger(name, log_file=None):
"""
配置日志，支持同时输出到控制台和文件
统一存放在根目录的 Logs 文件夹下
配置日志，支持同时输出到控制台和文件。
使用供应商代号作为父级 Logger，所有子 Logger 继承其 Handler，
并通过 propagate=False 避免与根 Logger 重复。
"""
if log_file is None:
# 获取项目根目录 (aiData)
root_dir = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
log_dir = os.path.join(root_dir, "Logs")
if not os.path.exists(log_dir):
os.makedirs(log_dir)
# 统一使用供应商代号作为文件名
supplier_code = os.path.basename(os.path.dirname(os.path.abspath(__file__)))
log_file = os.path.join(log_dir, f"{supplier_code}.log")
# 1. 获取供应商代号 (如 AiTeJiYiChong)
supplier_code = os.path.basename(os.path.dirname(os.path.abspath(__file__)))
# 2. 获取父级 Logger 并配置
parent_logger = logging.getLogger(supplier_code)
parent_logger.setLevel(logging.INFO)
parent_logger.propagate = False # 禁止向上传递给 root logger，防止重复
logger = logging.getLogger(name)
logger.setLevel(logging.INFO)
if not parent_logger.handlers:
if log_file is None:
# 获取项目根目录 (aiData)
root_dir = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
log_dir = os.path.join(root_dir, "Logs")
if not os.path.exists(log_dir):
os.makedirs(log_dir)
log_file = os.path.join(log_dir, f"{supplier_code}.log")
# 如果已经有 handler，不再添加
if logger.handlers:
return logger
formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')
formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')
# 控制台 Handler
ch = logging.StreamHandler()
ch.setFormatter(formatter)
parent_logger.addHandler(ch)
# 控制台 Handler
ch = logging.StreamHandler()
ch.setFormatter(formatter)
logger.addHandler(ch)
# 文件 Handler
fh = logging.FileHandler(log_file, encoding='utf-8')
fh.setFormatter(formatter)
parent_logger.addHandler(fh)
# 文件 Handler
fh = logging.FileHandler(log_file, encoding='utf-8')
fh.setFormatter(formatter)
logger.addHandler(fh)
return logger
# 3. 返回子 Logger
if name == supplier_code:
return parent_logger
return logging.getLogger(f"{supplier_code}.{name}")
def find_template_coords(img_path, template_path, threshold=0.8):

0
Apps/TelaiDian/Config/Setting.py → Apps/TeLaiDian/Config/Setting.py
View File

0
Apps/TelaiDian/Config/__init__.py → Apps/TeLaiDian/Config/__init__.py
View File

18
Apps/TelaiDian/Crawler.py → Apps/TeLaiDian/Crawler.py
View File

@@ -4,10 +4,10 @@ import logging
import os
import sys
import time
from Apps.TelaiDian.Kit import take_screenshot, get_image_content_md5, clean_station_name, setup_logger
from Apps.TelaiDian.ReadImageKit import ReadImageKit
from Apps.TelaiDian.Service import TelaiDianService
from Apps.TelaiDian.Config.Setting import (
from Apps.TeLaiDian.Kit import take_screenshot, get_image_content_md5, clean_station_name, setup_logger
from Apps.TeLaiDian.ReadImageKit import ReadImageKit
from Apps.TeLaiDian.Service import TeLaiDianService
from Apps.TeLaiDian.Config.Setting import (
SCROLL_DISTANCE_RATIO, WAIT_AFTER_SCROLL, MAX_STATIONS_COUNT,
SAFE_EXCLUDE_RATIO, BOTTOM_SAFE_EXCLUDE_RATIO, WAIT_DETAIL_PAGE_LOAD,
WAIT_BACK_TO_LIST
@@ -21,11 +21,11 @@ if project_root not in sys.path:
sys.path.append(project_root)
# 初始化日志
logger = setup_logger("TelaiDianCrawler")
logger = setup_logger("TeLaiDianCrawler")
class TelaiDianCrawler(BaseCrawler):
class TeLaiDianCrawler(BaseCrawler):
def __init__(self, service=None):
super().__init__(service or TelaiDianService())
super().__init__(service or TeLaiDianService())
self.read_image_kit = ReadImageKit()
async def start(self):
@@ -44,7 +44,7 @@ class TelaiDianCrawler(BaseCrawler):
"""
打开特来电小程序
"""
from Apps.TelaiDian import Opener
from Apps.TeLaiDian import Opener
return await Opener.open_mini_program()
async def crawl_list_logic(self, d):
@@ -134,7 +134,7 @@ class TelaiDianCrawler(BaseCrawler):
pass
async def main(service=None):
crawler = TelaiDianCrawler(service=service)
crawler = TeLaiDianCrawler(service=service)
await crawler.start()
if __name__ == "__main__":

0
Apps/TelaiDian/Doc/说明.md → Apps/TeLaiDian/Doc/说明.md
View File

0
Apps/TelaiDian/Images/1、列表页.jpg → Apps/TeLaiDian/Images/1、列表页.jpg
View File

Side by Side Swipe Overlay

Before

Width:  |  Height:  |  Size: 132 KiB

After

Width:  |  Height:  |  Size: 132 KiB

0
Apps/TelaiDian/Images/2、二级地址页.jpg → Apps/TeLaiDian/Images/2、二级地址页.jpg
View File

Side by Side Swipe Overlay

Before

Width:  |  Height:  |  Size: 144 KiB

After

Width:  |  Height:  |  Size: 144 KiB

0
Apps/TelaiDian/Images/3、二级向下滚动看到价格信息.jpg → Apps/TeLaiDian/Images/3、二级向下滚动看到价格信息.jpg
View File

Side by Side Swipe Overlay

Before

Width:  |  Height:  |  Size: 103 KiB

After

Width:  |  Height:  |  Size: 103 KiB

0
Apps/TelaiDian/Images/4、分时段收费信息.jpg → Apps/TeLaiDian/Images/4、分时段收费信息.jpg
View File

Side by Side Swipe Overlay

Before

Width:  |  Height:  |  Size: 82 KiB

After

Width:  |  Height:  |  Size: 82 KiB

58
Apps/TelaiDian/Kit.py → Apps/TeLaiDian/Kit.py
View File

@@ -195,36 +195,40 @@ def clear_temp_dir(save_dir=None):
def setup_logger(name, log_file=None):
"""
配置日志，支持同时输出到控制台和文件
统一存放在根目录的 Logs 文件夹下
配置日志，支持同时输出到控制台和文件。
使用供应商代号作为父级 Logger，所有子 Logger 继承其 Handler，
并通过 propagate=False 避免与根 Logger 重复。
"""
if log_file is None:
# 获取项目根目录 (aiData)
root_dir = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
log_dir = os.path.join(root_dir, "Logs")
if not os.path.exists(log_dir):
os.makedirs(log_dir)
# 统一使用供应商代号作为文件名
supplier_code = os.path.basename(os.path.dirname(os.path.abspath(__file__)))
log_file = os.path.join(log_dir, f"{supplier_code}.log")
# 1. 获取供应商代号 (如 TelaiDian)
supplier_code = os.path.basename(os.path.dirname(os.path.abspath(__file__)))
# 2. 获取父级 Logger 并配置
parent_logger = logging.getLogger(supplier_code)
parent_logger.setLevel(logging.INFO)
parent_logger.propagate = False # 禁止向上传递给 root logger，防止重复
logger = logging.getLogger(name)
logger.setLevel(logging.INFO)
if not parent_logger.handlers:
if log_file is None:
# 获取项目根目录 (aiData)
root_dir = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
log_dir = os.path.join(root_dir, "Logs")
if not os.path.exists(log_dir):
os.makedirs(log_dir)
log_file = os.path.join(log_dir, f"{supplier_code}.log")
# 如果已经有 handler，不再添加
if logger.handlers:
return logger
formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')
formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')
# 控制台 Handler
ch = logging.StreamHandler()
ch.setFormatter(formatter)
parent_logger.addHandler(ch)
# 控制台 Handler
ch = logging.StreamHandler()
ch.setFormatter(formatter)
logger.addHandler(ch)
# 文件 Handler
fh = logging.FileHandler(log_file, encoding='utf-8')
fh.setFormatter(formatter)
parent_logger.addHandler(fh)
# 文件 Handler
fh = logging.FileHandler(log_file, encoding='utf-8')
fh.setFormatter(formatter)
logger.addHandler(fh)
return logger
# 3. 返回子 Logger (如 TelaiDian.T4_TelaiDian)
if name == supplier_code:
return parent_logger
return logging.getLogger(f"{supplier_code}.{name}")

4
Apps/TelaiDian/Opener.py → Apps/TeLaiDian/Opener.py
View File

@@ -4,10 +4,10 @@ import logging
import os
import time
import uiautomator2 as u2
from Apps.TelaiDian.Kit import take_screenshot, setup_logger
from Apps.TeLaiDian.Kit import take_screenshot, setup_logger
# 初始化日志
logger = setup_logger("OpenTelaiDian")
logger = setup_logger("OpenTeLaiDian")
async def open_mini_program():
"""

4
Apps/TelaiDian/ReadImageKit.py → Apps/TeLaiDian/ReadImageKit.py
View File

@@ -10,8 +10,8 @@ if project_root not in sys.path:
sys.path.append(project_root)
from Util.VLMKit import VLMKit
from Apps.TelaiDian.Kit import draw_rectangles, detect_cards_cv, setup_logger
from Apps.TelaiDian.Config.Setting import SAFE_EXCLUDE_RATIO, BOTTOM_SAFE_EXCLUDE_RATIO
from Apps.TeLaiDian.Kit import draw_rectangles, detect_cards_cv, setup_logger
from Apps.TeLaiDian.Config.Setting import SAFE_EXCLUDE_RATIO, BOTTOM_SAFE_EXCLUDE_RATIO
# 初始化日志
logger = setup_logger("ReadImageKit")

6
Apps/TelaiDian/Service.py → Apps/TeLaiDian/Service.py
View File

@@ -16,12 +16,12 @@ from Config.Config import DB_URL
from Model.StationProfile import StationProfile
from Model.StationStatus import StationStatus
from Model.StationPriceSchedule import StationPriceSchedule
from Apps.TelaiDian.Kit import setup_logger
from Apps.TeLaiDian.Kit import setup_logger
# 初始化日志
logger = setup_logger("TelaiDianService")
logger = setup_logger("TeLaiDianService")
class TelaiDianService:
class TeLaiDianService:
def __init__(self):
self.db = Db(db_url=DB_URL)
self.station_profile_model = StationProfile()

0
Apps/TelaiDian/Template/jgxx.jpg → Apps/TeLaiDian/Template/jgxx.jpg
View File

Side by Side Swipe Overlay

Before

Width:  |  Height:  |  Size: 6.7 KiB

After

Width:  |  Height:  |  Size: 6.7 KiB

0
Apps/TelaiDian/__init__.py → Apps/TeLaiDian/__init__.py
View File

58
Apps/XinDianTu/Kit.py
View File

@@ -248,39 +248,43 @@ def detect_rabbit_ad_close(image_path, debug_dir=None):
def setup_logger(name, log_file=None):
"""
配置日志，支持同时输出到控制台和文件
统一存放在根目录的 Logs 文件夹下
配置日志，支持同时输出到控制台和文件。
使用供应商代号作为父级 Logger，所有子 Logger 继承其 Handler，
并通过 propagate=False 避免与根 Logger 重复。
"""
if log_file is None:
# 获取项目根目录 (aiData)
root_dir = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
log_dir = os.path.join(root_dir, "Logs")
if not os.path.exists(log_dir):
os.makedirs(log_dir)
# 统一使用供应商代号作为文件名
supplier_code = os.path.basename(os.path.dirname(os.path.abspath(__file__)))
log_file = os.path.join(log_dir, f"{supplier_code}.log")
# 1. 获取供应商代号 (如 XinDianTu)
supplier_code = os.path.basename(os.path.dirname(os.path.abspath(__file__)))
# 2. 获取父级 Logger 并配置
parent_logger = logging.getLogger(supplier_code)
parent_logger.setLevel(logging.INFO)
parent_logger.propagate = False # 禁止向上传递给 root logger，防止重复
logger = logging.getLogger(name)
logger.setLevel(logging.INFO)
if not parent_logger.handlers:
if log_file is None:
# 获取项目根目录 (aiData)
root_dir = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
log_dir = os.path.join(root_dir, "Logs")
if not os.path.exists(log_dir):
os.makedirs(log_dir)
log_file = os.path.join(log_dir, f"{supplier_code}.log")
# 如果已经有 handler，不再添加
if logger.handlers:
return logger
formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')
formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')
# 控制台 Handler
ch = logging.StreamHandler()
ch.setFormatter(formatter)
parent_logger.addHandler(ch)
# 控制台 Handler
ch = logging.StreamHandler()
ch.setFormatter(formatter)
logger.addHandler(ch)
# 文件 Handler
fh = logging.FileHandler(log_file, encoding='utf-8')
fh.setFormatter(formatter)
parent_logger.addHandler(fh)
# 文件 Handler
fh = logging.FileHandler(log_file, encoding='utf-8')
fh.setFormatter(formatter)
logger.addHandler(fh)
return logger
# 3. 返回子 Logger
if name == supplier_code:
return parent_logger
return logging.getLogger(f"{supplier_code}.{name}")
def detect_black_agree_button(image_path, debug_dir=None):

58
Apps/YeLiTe/Kit.py
View File

@@ -272,36 +272,40 @@ def draw_rectangles(image_path, points, output_path=None):
def setup_logger(name, log_file=None):
"""
配置日志，支持同时输出到控制台和文件
统一存放在根目录的 Logs 文件夹下
配置日志，支持同时输出到控制台和文件。
使用供应商代号作为父级 Logger，所有子 Logger 继承其 Handler，
并通过 propagate=False 避免与根 Logger 重复。
"""
if log_file is None:
# 获取项目根目录 (aiData)
root_dir = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
log_dir = os.path.join(root_dir, "Logs")
if not os.path.exists(log_dir):
os.makedirs(log_dir)
# 统一使用供应商代号作为文件名
supplier_code = os.path.basename(os.path.dirname(os.path.abspath(__file__)))
log_file = os.path.join(log_dir, f"{supplier_code}.log")
# 1. 获取供应商代号 (如 YeLiTe)
supplier_code = os.path.basename(os.path.dirname(os.path.abspath(__file__)))
# 2. 获取父级 Logger 并配置
parent_logger = logging.getLogger(supplier_code)
parent_logger.setLevel(logging.INFO)
parent_logger.propagate = False # 禁止向上传递给 root logger，防止重复
logger = logging.getLogger(name)
logger.setLevel(logging.INFO)
if not parent_logger.handlers:
if log_file is None:
# 获取项目根目录 (aiData)
root_dir = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
log_dir = os.path.join(root_dir, "Logs")
if not os.path.exists(log_dir):
os.makedirs(log_dir)
log_file = os.path.join(log_dir, f"{supplier_code}.log")
# 如果已经有 handler，不再添加
if logger.handlers:
return logger
formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')
formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')
# 控制台 Handler
ch = logging.StreamHandler()
ch.setFormatter(formatter)
parent_logger.addHandler(ch)
# 控制台 Handler
ch = logging.StreamHandler()
ch.setFormatter(formatter)
logger.addHandler(ch)
# 文件 Handler
fh = logging.FileHandler(log_file, encoding='utf-8')
fh.setFormatter(formatter)
parent_logger.addHandler(fh)
# 文件 Handler
fh = logging.FileHandler(log_file, encoding='utf-8')
fh.setFormatter(formatter)
logger.addHandler(fh)
return logger
# 3. 返回子 Logger
if name == supplier_code:
return parent_logger
return logging.getLogger(f"{supplier_code}.{name}")

8
T4_TelaiDian.py → T4_TeLaiDian.py
View File

@@ -16,10 +16,10 @@ logging.basicConfig(
)
try:
from Apps.TelaiDian.Service import TelaiDianService
from Apps.TelaiDian import Opener, Crawler, Kit
from Apps.TeLaiDian.Service import TeLaiDianService
from Apps.TeLaiDian import Opener, Crawler, Kit
# 初始化日志文件记录
logger = Kit.setup_logger("T4_TelaiDian")
logger = Kit.setup_logger("T4_TeLaiDian")
except KeyboardInterrupt:
logging.info("\n🛑 用户在初始化阶段手动停止了程序。")
sys.exit(0)
@@ -32,7 +32,7 @@ async def run_process():
# 步骤 0: 初始化基础服务
logger.info(">>> 步骤 0: 初始化基础服务 (数据库连接)...")
service = TelaiDianService()
service = TeLaiDianService()
await service.init_db()
try: