| 1 | import re |
| 2 | from typing import Optional |
| 3 | from urllib.parse import urlparse |
| 4 | |
| 5 | |
| 6 | def validate_url(url: str) -> bool: |
| 7 | try: |
| 8 | result = urlparse(url) |
| 9 | return all([result.scheme, result.netloc]) |
| 10 | except Exception: |
| 11 | return False |
| 12 | |
| 13 | |
| 14 | def sanitize_filename(filename: str, max_length: int = 80) -> str: |
| 15 | # 换行符 → 空格 |
| 16 | filename = filename.replace("\n", " ").replace("\r", " ") |
| 17 | # Windows 非法字符 + #，逗号 → 下划线 |
| 18 | filename = re.sub(r'[<>:"/\\|?*#\x00-\x1f]', "_", filename) |
| 19 | # 连续下划线 → 单个下划线（保留空格，不再把空格折叠成下划线） |
| 20 | filename = re.sub(r"_+", "_", filename) |
| 21 | # 连续空格 → 单个空格 |
| 22 | filename = re.sub(r" +", " ", filename) |
| 23 | # 去首尾 |
| 24 | filename = filename.strip("._- ") |
| 25 | |
| 26 | if len(filename) > max_length: |
| 27 | filename = filename[:max_length].rstrip("._- ") |
| 28 | |
| 29 | return filename or "untitled" |
| 30 | |
| 31 | |
| 32 | SHORT_URL_HOSTS = ( |
| 33 | "v.douyin.com", |
| 34 | "v.iesdouyin.com", |
| 35 | "iesdouyin.com", |
| 36 | ) |
| 37 | |
| 38 | |
| 39 | def is_short_url(url: str) -> bool: |
| 40 | """判断是否为需要预先解析的短链。""" |
| 41 | if not url: |
| 42 | return False |
| 43 | # 允许用户粘贴不带 scheme 的短链（例如直接从 App 复制） |
| 44 | candidate = url.strip() |
| 45 | lowered = candidate.lower() |
| 46 | for scheme in ("https://", "http://"): |
| 47 | if lowered.startswith(scheme): |
| 48 | lowered = lowered[len(scheme) :] |
| 49 | break |
| 50 | for host in SHORT_URL_HOSTS: |
| 51 | if lowered.startswith(f"{host}/") or lowered == host: |
| 52 | return True |
| 53 | return False |
| 54 | |
| 55 | |
| 56 | def normalize_short_url(url: str) -> str: |
| 57 | """确保短链带 https:// 前缀，便于传给 aiohttp。""" |
| 58 | stripped = (url or "").strip() |
| 59 | if stripped.lower().startswith(("http://", "https://")): |
| 60 | return stripped |
| 61 | return f"https://{stripped}" |
| 62 | |
| 63 | |
| 64 | def parse_url_type(url: str) -> Optional[str]: |
| 65 | # 短链在调用方（CLI/调度层）统一先解析为真实 URL 后再判断类型； |
| 66 | # 若仍是短链，返回 'short' 明确提示需要解析，而不是错误地全部落到 'video'。 |
| 67 | if is_short_url(url): |
| 68 | return "short" |
| 69 | |
| 70 | parsed = urlparse(url) |
| 71 | host = (parsed.netloc or "").lower() |
| 72 | path = parsed.path |
| 73 | |
| 74 | # live.douyin.com/{room_id} — 直播间专用子域，path 仅有一段数字。 |
| 75 | if host.startswith("live.douyin.com"): |
| 76 | return "live" |
| 77 | |
| 78 | if "/video/" in path: |
| 79 | return "video" |
| 80 | if "/user/" in path: |
| 81 | return "user" |
| 82 | if "/note/" in path or "/gallery/" in path or "/slides/" in path: |
| 83 | return "gallery" |
| 84 | if "/collection/" in path or "/mix/" in path: |
| 85 | return "collection" |
| 86 | if "/music/" in path: |
| 87 | return "music" |
| 88 | if "/live/" in path or "/follow/live/" in path: |
| 89 | return "live" |
| 90 | return None |
| 91 |