上线文新百度ai

上线图片下载
电商数据豆包逻辑修复

aidso_geo/config/base_config.py

@@ -105,11 +105,11 @@ class BaseConfig:
             "storage_path": lambda tid: f"geo/{tid}/KIMI/original.text"
         },
         PlatformType.WENXINYIYAN.value: {
-            "url": f"{base_url}wenxinyiyan",
+            "url": f"{base_url}baiduai",
             "storage_path": lambda tid: f"geo/{tid}/WXYY/original.text"
         },
         PlatformType.BAIDUAI.value: {
-            "url": f"{base_url}baiduai",
+            "url": f"{base_url}baikan",
             "storage_path": lambda tid: f"geo/{tid}/BDAI/original.text"
         },
         PlatformType.DOUYINAI.value: {

aidso_geo/core/third_interface.py

-#!/usr/bin/env python3
-# -*- coding: utf-8 -*-
-
-import time
-from datetime import datetime
-
-from aidso_geo.config.base_config import init_redis8
-from aidso_geo.utils import bh_utils
-
-
-redis_client = init_redis8()
-
-
-def safe_int(value, default=0):
-    try:
-        if value is None:
-            return default
-        return int(value)
-    except Exception:
-        return default
-
-
-def get_active_channels():
-    """
-    查询所有启用中的第三方 channel
-    """
-    sql = """
-        select channel, daily_limit, total_limit
-        from geo_third_token
-        where status = 1
-    """
-    return bh_utils.query_data(sql) or []
-
-
-def sync_third_usage_once():
-    """
-    每小时同步第三方 token 用量快照。
-
-    读取：
-    1. Redis hash third_geo_daily：field = channel + YYYYMMDD
-    2. Redis hash third_geo_total：field = channel
-
-    写入：
-    geo_third_usage_snapshot
-    """
-    now = datetime.now()
-    today_pt = now.strftime("%Y%m%d")
-    sync_time = now.strftime("%Y-%m-%d %H:%M:%S")
-    insertime = int(time.time())
-
-    try:
-        channels = get_active_channels()
-        if not channels:
-            logger.info(f"[{sync_time}] sync_third_usage_once 无启用中的 channel")
-            return
-
-        # 一次性拉 Redis hash，避免用户多时频繁 hget
-        daily_map = redis_client.hgetall("third_geo_daily") or {}
-        total_map = redis_client.hgetall("third_geo_total") or {}
-
-        insert_rows = []
-
-        for item in channels:
-            channel = str(item.get("channel", "")).strip()
-            if not channel:
-                continue
-
-            daily_limit = safe_int(item.get("daily_limit"), 0)
-            total_limit = safe_int(item.get("total_limit"), 0)
-
-            daily_key = f"{channel}{today_pt}"
-
-            today_used = safe_int(daily_map.get(daily_key), 0)
-            total_used = safe_int(total_map.get(channel), 0)
-
-            insert_rows.append({
-                "channel": channel,
-                "pt": today_pt,
-                "daily_limit": daily_limit,
-                "total_limit": total_limit,
-                "today_used": today_used,
-                "total_used": total_used,
-                "today_remain": max(daily_limit - today_used, 0),
-                "total_remain": max(total_limit - total_used, 0),
-                "sync_time": sync_time,
-                "insertime": insertime
-            })
-
-        if not insert_rows:
-            logger.info(f"[{sync_time}] sync_third_usage_once 无可写入数据")
-            return
-
-        ok = bh_utils.insert_data("geo_third_usage_snapshot", insert_rows)
-        if ok:
-            logger.info(f"[{sync_time}] sync_third_usage_once 同步成功 rows={len(insert_rows)}")
-        else:
-            logger.error(f"[{sync_time}] sync_third_usage_once 同步失败 rows={len(insert_rows)}")
-
-    except Exception as e:
-        logger.exception(f"[{sync_time}] sync_third_usage_once 执行异常: {e}")
-
-
-if __name__ == '__main__':
-    sync_third_usage_once()
\ No newline at end of file

aidso_geo/models/doubao_android_data_process.py

@@ -214,7 +214,7 @@ def doubao_mobile_process_original_data(task_data):
                                                                                                     'jump_url': url,
                                                                                                 }
                                                                                                 media_list.append(product)
-                                                                                                dy_eco_list.extend(media_list)
+                                                                                                dy_eco_list.append(product)
                                                                                     poi_data = widget_data_data.get(
                                                                                         'poi_list')
 
@@ -310,6 +310,7 @@ def doubao_mobile_process_original_data(task_data):
                                                suggestions, rich_media_block)
         return (file_path, search_keyword, url_list, think_content, response_content, suggestions)
     except Exception as e:
+
         parts = file_path.split('/')
         platform = parts[2]
         task_id = parts[1]
@@ -342,7 +343,7 @@ if __name__ == '__main__':
     # file_path = 'geo/c7eb465e-f385-4aa2-89c4-a7cf11897f45/KIMI/1(1).txt'
 
     data_list = bh_utils.query_data(
-        f"select * from geo_commit_task where platform = 'DOUBA' and insertime>1783008000 and type !='success'")
+        f"select * from geo_commit_task where req_id = '000804db-804b-428d-b5d8-f3d0b717bc3b'")
 
 
 

aidso_geo/models/eco_data_process.py

@@ -90,3 +90,5 @@ def save_eco_data_to_bh(data,eco_type,eco_list):
         if eco_type == 'dyeco':
             eco_result =doubao_android_process_douyin_eco(data,eco_list)
     bh_utils.insert_data('geo_eco_data',eco_result)
+
+

aidso_geo/models/process.py

@@ -13,6 +13,7 @@ from enum import Enum
 import json
 from concurrent.futures import ThreadPoolExecutor, as_completed
 
+from aidso_geo.models.doubao_android_data_process import doubao_mobile_process_original_data
 from aidso_geo.utils.tos_utils import put_string_to_tos, check_file_in_tos, get_tos_file_size
 BASE_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
 sys.path.append(BASE_DIR)
@@ -1635,9 +1636,9 @@ def platform_process(data):
         'DP': deepseek_data_process.deepseek_process_original_data,
         'TXYB': yuanbao_data_process.yuanbao_process_original_data,
         'KIMI': kimi_data_process.kimi_process_original_data,
-        'WXYY': wenxin_data_process.wenxin_process_original_data,
+        'WXYY': baiduai_data_process.baiduai_process_original_data,
         'TYQW': qianwen_data_process.qianwen_process_original_data,
-        'BDAI': baiduai_data_process.baiduai_process_original_data,
+        'BDAI': baidu_baikan_data_process.baidu_baikan_data_process_original_data,
         'DPA': deepseek_android_data_process.deepseek_android_process_original_data,
         'DOUBA': doubao_android_data_process.doubao_mobile_process_original_data,
         'DYAI': douyinai_data_process.douyin_ai_process_original_data,
@@ -1954,7 +1955,7 @@ def run_data(PAGE_SIZE,MAX_WORKERS):
 
 if __name__ == '__main__':
 
-    data_list = bh_utils.query_data(f"select * from geo_commit_task where status = 'ING'")
+    data_list = bh_utils.query_data(f"select * from geo_commit_task where status ='ING'")
     # data_list = bh_utils.query_data(query_sql)
     # print(data_list)
     # # #

aidso_geo/models/qianwen_android_data_process.py

@@ -366,12 +366,10 @@ if __name__ == '__main__':
 
 
     # data_list = bh_utils.query_data(f"select * from geo_commit_task  where taskId = '0001e01d-a8ea-405e-acef-93e4f55abbff' and  platform  = 'TYQWA'")
-    data_list = bh_utils.query_data(f"select * from geo_commit_task where platform = 'TYQWA' and insertime>1783008000 and type !='success'")
+    data_list = bh_utils.query_data(f"select * from geo_commit_task where taskId = '374593fc-ac18-4eb3-9815-866704ce02ca' and platform = 'TYQWA'")
 
     def handle_item(i):
-        task_id = i.get('taskId')
-        if not get_string_from_tos(f"geo/{task_id}/TYQWA/quote.txt"):
-            logger.success(f"{task_id}处理完成")
+
         if i.get('comWordsMap'):
             i['comWordsMap'] = json.loads(i.get('comWordsMap'))
         if i.get('brandWords'):

aidso_geo/models/spider_save_tos.py

@@ -474,6 +474,8 @@ def save_data_to_tos(target_dir, content, file_name):
             content = qianwen_android_process_rich_media(content)
         if platform =='BDAI':
             content = baidu_ai_rich_media(content)
+        if platform =='WXYY':
+            content = baidu_ai_rich_media(content)
         if platform =='TYQW':
             content = qianwen_process_rich_media(content)
         if platform == 'TXYBA':
@@ -493,9 +495,9 @@ def save_data_to_tos(target_dir, content, file_name):
         elif platform == 'KIMI':
             content = kimi_process_quote(content)
         elif platform == 'WXYY':
-            content = wenxin_process_quote(content)
-        elif platform == 'BDAI':
             content = baiduai_process_quote(content)
+        elif platform == 'BDAI':
+            content = baikan_process_quote(content)
         elif platform == 'DPA':
             content = deepseek_android_process_quote(content)
         elif platform == 'DOUBA':

aidso_geo/utils/robot_utils.py

@@ -5,8 +5,8 @@ import json
 from loguru import logger
 from apscheduler.schedulers.blocking import BlockingScheduler
 
-from aidso_geo.config.base_config import init_redis
-from aidso_geo.utils import bh_utils
+from aidso_geo.config.base_config import init_redis, init_redis8
+from aidso_geo.utils import bh_utils,tos_utils
 
 owner_map = {
     "XHSA": "崔士豪",
@@ -247,6 +247,119 @@ def task_queue_backlog():
     except Exception as e:
         logger.exception(f"task_queue_backlog 执行失败: {e}")
 
+from datetime import datetime
+
+redis_client = init_redis8()
+
+
+def safe_int(value, default=0):
+    try:
+        if value is None:
+            return default
+        return int(value)
+    except Exception:
+        return default
+
+
+def get_active_channels():
+    """
+    查询所有启用中的第三方 channel
+    """
+    sql = """
+        select channel, daily_limit, total_limit
+        from geo_third_token
+        where status = 1
+    """
+    return bh_utils.query_data(sql) or []
+
+
+def sync_third_usage_once():
+    """
+    每小时同步第三方 token 用量快照。
+
+    读取：
+    1. Redis hash third_geo_daily：field = channel + YYYYMMDD
+    2. Redis hash third_geo_total：field = channel
+
+    写入：
+    geo_third_usage_snapshot
+    """
+    now = datetime.now()
+    today_pt = now.strftime("%Y%m%d")
+    sync_time = now.strftime("%Y-%m-%d %H:%M:%S")
+    insertime = int(time.time())
+
+    try:
+        channels = get_active_channels()
+        if not channels:
+            logger.info(f"[{sync_time}] sync_third_usage_once 无启用中的 channel")
+            return
+
+        # 一次性拉 Redis hash，避免用户多时频繁 hget
+        daily_map = redis_client.hgetall("third_geo_daily") or {}
+        total_map = redis_client.hgetall("third_geo_total") or {}
+
+        insert_rows = []
+
+        for item in channels:
+            channel = str(item.get("channel", "")).strip()
+            if not channel:
+                continue
+
+            daily_limit = safe_int(item.get("daily_limit"), 0)
+            total_limit = safe_int(item.get("total_limit"), 0)
+
+            daily_key = f"{channel}{today_pt}"
+
+            today_used = safe_int(daily_map.get(daily_key), 0)
+            total_used = safe_int(total_map.get(channel), 0)
+
+            insert_rows.append({
+                "channel": channel,
+                "pt": today_pt,
+                "daily_limit": daily_limit,
+                "total_limit": total_limit,
+                "today_used": today_used,
+                "total_used": total_used,
+                "today_remain": max(daily_limit - today_used, 0),
+                "total_remain": max(total_limit - total_used, 0),
+                "sync_time": sync_time,
+                "insertime": insertime
+            })
+
+        if not insert_rows:
+            logger.info(f"[{sync_time}] sync_third_usage_once 无可写入数据")
+            return
+
+        ok = bh_utils.insert_data("geo_third_usage_snapshot", insert_rows)
+        if ok:
+            logger.info(f"[{sync_time}] sync_third_usage_once 同步成功 rows={len(insert_rows)}")
+        else:
+            logger.error(f"[{sync_time}] sync_third_usage_once 同步失败 rows={len(insert_rows)}")
+
+    except Exception as e:
+        logger.exception(f"[{sync_time}] sync_third_usage_once 执行异常: {e}")
+
+
+def save_image():
+    try:
+        result = bh_utils.query_data("""
+            select distinct t1.eco_id, t1.eco_pic
+        from geo_eco_data t1
+                 left join geo_eco_image_data t2 on t1.eco_id = t2.eco_id
+        where t2.eco_id is null limit 10000
+            """)
+        bh_result = []
+        for r in result:
+            eco_pic = r.get("eco_pic")
+            eco_id = r.get("eco_id")
+            logger.info(f"{eco_id}处理成功")
+            tt = tos_utils.save_image_to_tos(r.get("eco_pic"), f'geo_image/{r.get("eco_id")}')
+            bh_result.append({"eco_id": eco_id, "eco_pic": eco_pic, "eco_pic_tos": tt})
+        bh_utils.insert_data("geo_eco_image_data", bh_result)
+    except Exception as e:
+        logger.exception(f"task_queue_backlog 执行失败: {e}")
+
 
 if __name__ == '__main__':
     logger.info("监控调度器启动")
@@ -261,7 +374,7 @@ if __name__ == '__main__':
         replace_existing=True
     )
 
-    # task_queue_backlog：每小时整点执行一次
+
     scheduler.add_job(
         task_queue_backlog,
         trigger='cron',
@@ -272,9 +385,32 @@ if __name__ == '__main__':
         replace_existing=True
     )
 
+
+    scheduler.add_job(
+        save_image,
+        trigger='cron',
+        minute=30,
+        id='save_image',
+        max_instances=1,
+        coalesce=True,
+        replace_existing=True
+    )
+
+    scheduler.add_job(
+        sync_third_usage_once,
+        trigger='cron',
+        minute=5,
+        id='sync_third_usage_once',
+        max_instances=1,
+        coalesce=True,
+        replace_existing=True
+    )
+
     logger.info(
         "定时任务注册完成："
         "fail_task_send_feishu(每6小时), "
         "task_queue_backlog(每小时)"
+        "save_image(每小时)"
+        "sync_third_usage_once(每小时)"
     )
     scheduler.start()

aidso_geo/utils/tos_utils.py

-import time
 import json
 import os, sys
-
 BASE_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
 sys.path.append(BASE_DIR)
 from aidso_geo.clients.tos_client import tos_client
 import tos
+import requests
+from io import BytesIO
+from PIL import Image
 
 
 def put_string_to_tos(object_key: str, content, retry_times: int = 3) -> bool:
@@ -79,8 +80,91 @@ def get_tos_file_size(object_key: str, retry_times: int = 3):
     return None
 
 
-if __name__ == '__main__':
-    path = f'geo/5ef9bbf4-b356-40af5d44fff30a/TXYB/original.text'
-    # put_string_to_tos('geo/1005/DP/aaaa.txt',"['2025年11月5日 天气']")
-    # print(check_file_in_tos('geo/1d38cb5e-0b99-49e8-8f68-13688079b719/DP/result.json'))
-    print(get_tos_file_size(path))
+
+
+
+
+def save_image_to_tos(image_url, tos_path=None):
+
+    client = tos_client.get_client()
+    bucket = tos_client.bucket_name
+
+    try:
+
+        # 下载图片
+        resp = requests.get(
+            image_url,
+            timeout=10,
+            headers={
+                "User-Agent": "Mozilla/5.0"
+            }
+        )
+
+
+        if resp.status_code != 200:
+
+            print(
+                f"图片下载失败:{image_url}, status={resp.status_code}"
+            )
+
+            return None
+
+
+
+        # 打开图片
+        image = Image.open(
+            BytesIO(resp.content)
+        )
+
+
+        # 转RGB，保证可以保存JPEG
+        image = image.convert("RGB")
+
+
+
+        # 转jpg二进制
+        buffer = BytesIO()
+
+        image.save(
+            buffer,
+            format="JPEG",
+            quality=95
+        )
+
+        buffer.seek(0)
+
+
+
+
+
+        tos_path = str(tos_path)
+
+        if not tos_path.endswith(".jpg"):
+
+            tos_path += ".jpg"
+
+
+
+        # 上传TOS
+        client.put_object(
+            bucket,
+            tos_path,
+            content=buffer.getvalue(),
+            content_type="image/jpeg"
+        )
+
+
+
+
+        return tos_path
+
+
+
+    except Exception as e:
+
+        print(
+            f"保存图片到TOS失败:{image_url}, error={e}"
+        )
+
+        return None
+