84 lines
3.1 KiB
Python
84 lines
3.1 KiB
Python
|
import asyncio
|
|||
|
|
from playwright.async_api import async_playwright
|
|||
|
|
import re
|
|||
|
|
import yaml
|
|||
|
|
|
|||
|
|
async def resolve_url_and_extract(short_url):
|
|||
|
|
print(f"正在解析链接: {short_url}")
|
|||
|
|
async with async_playwright() as p:
|
|||
|
|
# 使用 headless=True 和 禁用GPU
|
|||
|
|
browser = await p.chromium.launch(headless=True, args=['--disable-gpu'])
|
|||
|
|
# 模拟手机，有的跳转可能依赖UA
|
|||
|
|
device = p.devices['iPhone 13']
|
|||
|
|
context = await browser.new_context(**device)
|
|||
|
|
page = await context.new_page()
|
|||
|
|
|
|||
|
|
try:
|
|||
|
|
await page.goto(short_url)
|
|||
|
|
# 等待跳转完成
|
|||
|
|
await page.wait_for_load_state('networkidle')
|
|||
|
|
await asyncio.sleep(2) # 额外等待确保 URL 稳定
|
|||
|
|
|
|||
|
|
final_url = page.url
|
|||
|
|
print(f"最终链接: {final_url}")
|
|||
|
|
|
|||
|
|
item_id = ""
|
|||
|
|
shop_id = ""
|
|||
|
|
|
|||
|
|
# 尝试从 URL 提取
|
|||
|
|
# 常见的 param 是 itemID=xxx, shopId=xxx
|
|||
|
|
# 或者路径中 /item.html?itemID=...
|
|||
|
|
|
|||
|
|
item_id_match = re.search(r"[?&]itemID=(\d+)", final_url, re.IGNORECASE)
|
|||
|
|
if item_id_match:
|
|||
|
|
item_id = item_id_match.group(1)
|
|||
|
|
|
|||
|
|
shop_id_match = re.search(r"[?&]shopId=(\d+)", final_url, re.IGNORECASE)
|
|||
|
|
if shop_id_match:
|
|||
|
|
shop_id = shop_id_match.group(1)
|
|||
|
|
|
|||
|
|
# 如果 URL 里没有，尝试在页面内容里找（有时候是全局变量）
|
|||
|
|
if not item_id or not shop_id:
|
|||
|
|
content = await page.content()
|
|||
|
|
if not item_id:
|
|||
|
|
# 匹配 "itemID":"123123" 或 itemID = '123123'
|
|||
|
|
m = re.search(r'["\']?itemID["\']?\s*[:=]\s*["\']?(\d+)["\']?', content, re.IGNORECASE)
|
|||
|
|
if m: item_id = m.group(1)
|
|||
|
|
|
|||
|
|
if not shop_id:
|
|||
|
|
m = re.search(r'["\']?shopId["\']?\s*[:=]\s*["\']?(\d+)["\']?', content, re.IGNORECASE)
|
|||
|
|
if m: shop_id = m.group(1)
|
|||
|
|
|
|||
|
|
print(f"解析结果 -> itemID: {item_id}, shopId: {shop_id}")
|
|||
|
|
return final_url, item_id, shop_id
|
|||
|
|
|
|||
|
|
except Exception as e:
|
|||
|
|
print(f"解析出错: {e}")
|
|||
|
|
return None, None, None
|
|||
|
|
finally:
|
|||
|
|
await browser.close()
|
|||
|
|
|
|||
|
|
def update_config(url, item_id, shop_id):
|
|||
|
|
config_path = "config.yaml"
|
|||
|
|
with open(config_path, "r", encoding="utf-8") as f:
|
|||
|
|
config = yaml.safe_load(f)
|
|||
|
|
|
|||
|
|
config['target_url'] = url
|
|||
|
|
if item_id:
|
|||
|
|
config['item_id'] = item_id
|
|||
|
|
if shop_id:
|
|||
|
|
config['shop_id'] = shop_id
|
|||
|
|
|
|||
|
|
with open(config_path, "w", encoding="utf-8") as f:
|
|||
|
|
yaml.dump(config, f, allow_unicode=True, sort_keys=False)
|
|||
|
|
print("配置 config.yaml 已更新。")
|
|||
|
|
|
|||
|
|
if __name__ == "__main__":
|
|||
|
|
# URL to resolve
|
|||
|
|
target = "https://k.youshop10.com/cTO2VL6s?a=b&p=iphone&wfr=BuyercopyURL&share_relation=c03c72974993c056_1767112998_1"
|
|||
|
|
|
|||
|
|
final_url, item_id, shop_id = asyncio.run(resolve_url_and_extract(target))
|
|||
|
|
|
|||
|
|
if final_url:
|
|||
|
|
update_config(final_url, item_id, shop_id)
|