from __future__ import annotations from acquisition.unique_key import build_unique_key, parse_weixin_unique_key def test_build_unique_key_for_xiaohongshu_and_douyin(): assert ( build_unique_key("xiaohongshu", platform_item_id="685a477d00000000120211af") == "xhs:685a477d00000000120211af" ) assert ( build_unique_key("douyin", platform_item_id="7612592616106388138") == "dy:7612592616106388138" ) def test_build_unique_key_reads_raw_payload_fallbacks(): assert ( build_unique_key("xiaohongshu", raw_payload={"candidate": {"source_id": "xhs-1"}}) == "xhs:xhs-1" ) assert ( build_unique_key("douyin", raw_payload={"candidate": {"raw": {"id": "dy-1"}}}) == "dy:dy-1" ) def test_parse_weixin_unique_key_from_article_url(): url = ( "https://mp.weixin.qq.com/s?__biz=MzUxMDg4NDY1Ng==" "&mid=2247484898&idx=2&sn=5069e65919f414b96b7aec53671dc1b5" "&chksm=abc#rd" ) assert ( parse_weixin_unique_key(url) == "wx:MzUxMDg4NDY1Ng==:2247484898:2:5069e65919f414b96b7aec53671dc1b5" ) assert build_unique_key("weixin", url=url) == parse_weixin_unique_key(url) def test_weixin_unique_key_is_none_when_url_lacks_article_parts(): assert parse_weixin_unique_key("https://mp.weixin.qq.com/s/a") is None assert build_unique_key("weixin", platform_item_id="old-hash", url="https://mp.weixin.qq.com/s/a") is None