#!/usr/bin/env python3 # -*- coding: utf-8 -*- import datetime import html import json import os import re import time import uuid import urllib.parse import urllib.request WORKSPACE = os.path.dirname(os.path.abspath(__file__)) HOT_VIDEOS_PATH = os.path.join(WORKSPACE, "hot-videos.json") TRACK = "AI宠物带货赛道" MIN_DAILY_VIDEOS = 10 ALLOWED_PLATFORMS = {"抖音", "小红书", "B站"} BLOCKED_URL_PARTS = ("kuaishou.com", "gifshow.com") DOUYIN_PUBLIC_SEEDS = [ { "platform": "抖音", "title": "用豆包来做宠物带货视频简直不要太简单", "url": "https://www.douyin.com/video/7621187940072312115", "likes": 10000, "tags": ["带货", "豆包", "宠物", "AI", "自媒体"], "notes": "公开链接池保留的真实抖音视频；播放量与粉丝数需登录态补全。", }, { "platform": "抖音", "title": "最近爆火的ai宠物带货视频，原来这么高的？", "url": "https://www.douyin.com/video/7619026937466170286", "likes": 21, "tags": ["AI宠物", "AI带货", "AI工具", "轻智AI"], "notes": "公开链接池保留的真实抖音视频；播放量与粉丝数需登录态补全。", }, { "platform": "抖音", "title": "两分钟让你学会宠物带货", "url": "https://www.douyin.com/video/7656026524445935049", "tags": ["AI视频", "AI教程", "宠物带货", "治愈系动画"], "notes": "公开链接池保留的真实抖音视频；播放量与粉丝数需登录态补全。", }, ] SEARCH_QUERIES = [ "AI宠物带货", "宠物电商带货 AI视频", "AI宠物用品 带货视频", "Coze 宠物 电商 带货 视频", "宠物科普 带货 AI 教程", ] def now_str(): tz8 = datetime.timezone(datetime.timedelta(hours=8)) return datetime.datetime.now(tz8).strftime("%Y-%m-%d %H:%M:%S") def browser_headers(referer="https://search.bilibili.com/"): return { "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/126.0.0.0 Safari/537.36", "Referer": referer, "Origin": "https://search.bilibili.com", "Accept": "application/json, text/plain, */*", "Accept-Language": "zh-CN,zh;q=0.9,en;q=0.8", "Cookie": f"buvid3={uuid.uuid4().hex.upper()}infoc; b_nut={int(time.time())}; CURRENT_FNVAL=4048;", } def fetch_json(url, timeout=15, referer="https://search.bilibili.com/"): req = urllib.request.Request( url, headers=browser_headers(referer), ) with urllib.request.urlopen(req, timeout=timeout) as resp: return json.loads(resp.read().decode("utf-8", "ignore")) def clean_title(title): title = re.sub(r"]*>|", "", title or "", flags=re.I) title = re.sub(r"<[^>]+>", "", title) return html.unescape(re.sub(r"\s+", " ", title)).strip() def normalize_number(value): if value in (None, "", "-"): return None if isinstance(value, (int, float)): return int(value) text = str(value).strip().lower() try: if text.endswith("万"): return int(float(text[:-1]) * 10000) if text.endswith("亿"): return int(float(text[:-1]) * 100000000) return int(float(text)) except Exception: return None def get_bilibili_fans(mid): if not mid: return None try: url = "https://api.bilibili.com/x/relation/stat?" + urllib.parse.urlencode({"vmid": mid}) data = fetch_json(url, timeout=8, referer="https://www.bilibili.com/") return normalize_number(((data.get("data") or {}).get("follower"))) except Exception: return None def extract_tags(raw): if isinstance(raw, list): tags = raw else: tags = re.split(r"[,，#\s]+", str(raw or "")) tags = [clean_title(t) for t in tags if clean_title(t)] base = ["AI宠物带货"] for tag in tags: if tag not in base: base.append(tag) if len(base) >= 5: break return base def is_track_relevant(text): text = text or "" has_ai = any(k in text for k in ("AI", "ai", "Coze", "扣子", "豆包", "智能体", "AIGC")) has_pet = any(k in text for k in ("宠物", "萌宠", "猫", "狗", "养宠")) has_commerce = any(k in text for k in ("带货", "电商", "商品", "用品", "零食", "橱窗", "视频")) return has_ai and has_pet and has_commerce def collect_bilibili(): results = [] seen = set() for query in SEARCH_QUERIES: url = "https://api.bilibili.com/x/web-interface/search/type?" + urllib.parse.urlencode( {"search_type": "video", "keyword": query, "page": 1, "page_size": 20} ) data = fetch_json(url, referer="https://search.bilibili.com/all?keyword=" + urllib.parse.quote(query)) if data.get("code") != 0: continue for item in (data.get("data") or {}).get("result") or []: title = clean_title(item.get("title")) arcurl = item.get("arcurl") or "" bvid = item.get("bvid") or "" url = f"https://www.bilibili.com/video/{bvid}/" if bvid else arcurl.replace("http://", "https://") if not title or "bilibili.com" not in url or url in seen: continue text = title + " " + str(item.get("tag") or "") if not is_track_relevant(text): continue plays = normalize_number(item.get("play")) likes = normalize_number(item.get("like")) fans = get_bilibili_fans(item.get("mid")) ratio = round(plays / fans, 2) if plays and fans else None results.append( { "platform": "B站", "title": title, "url": url, "plays": plays, "playCount": plays, "likes": likes, "likeCount": likes, "fans": fans, "followerCount": fans, "playFanRatio": ratio, "tags": extract_tags(item.get("tag")), "author": item.get("author") or "", "capturedAt": now_str(), "notes": "B站公开搜索接口抓取。", } ) seen.add(url) return results def collect_douyin_public(): captured = now_str() return [{**item, "plays": item.get("plays"), "playCount": item.get("plays"), "fans": item.get("fans"), "followerCount": item.get("fans"), "playFanRatio": None, "capturedAt": captured} for item in DOUYIN_PUBLIC_SEEDS] def dedupe(videos): output = [] seen_urls = set() seen_titles = set() for video in videos: platform = video.get("platform") url = str(video.get("url") or "") title = clean_title(video.get("title")) if platform not in ALLOWED_PLATFORMS: continue if any(part in url.lower() for part in BLOCKED_URL_PARTS): continue key_title = re.sub(r"\W+", "", title.lower())[:36] if not url or not title or url in seen_urls or key_title in seen_titles: continue video["title"] = title output.append(video) seen_urls.add(url) seen_titles.add(key_title) return output def sort_key(video): ratio = video.get("playFanRatio") if isinstance(ratio, (int, float)): return (1, ratio, normalize_number(video.get("plays")) or 0, normalize_number(video.get("likes")) or 0) return (0, 0, normalize_number(video.get("plays")) or 0, normalize_number(video.get("likes")) or 0) def main(): notes = [] videos = [] try: videos.extend(collect_bilibili()) except Exception as e: notes.append(f"B站公开接口抓取失败：{e}") try: videos.extend(collect_douyin_public()) notes.append("抖音公开网页/接口多数需要登录态，已保留真实公开链接，指标缺失不强行填充。") except Exception as e: notes.append(f"抖音公开链接池写入失败：{e}") notes.append("小红书公开视频搜索结果稳定性较差，未拿到可验证真实链接时不写入；本次由 B站公开结果补足到 10 条。") videos = dedupe(videos) videos.sort(key=sort_key, reverse=True) selected = videos[:MIN_DAILY_VIDEOS] if len(selected) < MIN_DAILY_VIDEOS: notes.append(f"最终只得到 {len(selected)} 条有效视频，未达到 10 条。") payload = { "updatedAt": now_str(), "source": "daily automation", "track": TRACK, "minDailyVideos": MIN_DAILY_VIDEOS, "videos": selected, "notes": "；".join(notes), } with open(HOT_VIDEOS_PATH, "w", encoding="utf-8") as f: json.dump(payload, f, ensure_ascii=False, indent=2) print(json.dumps({"ok": len(selected) >= MIN_DAILY_VIDEOS, "count": len(selected), "path": HOT_VIDEOS_PATH, "notes": payload["notes"]}, ensure_ascii=False)) if __name__ == "__main__": main()