test_unique_key.py 1.4 KB

1234567891011121314151617181920212223242526272829303132333435363738394041424344
  1. from __future__ import annotations
  2. from acquisition.unique_key import build_unique_key, parse_weixin_unique_key
  3. def test_build_unique_key_for_xiaohongshu_and_douyin():
  4. assert (
  5. build_unique_key("xiaohongshu", platform_item_id="685a477d00000000120211af")
  6. == "xhs:685a477d00000000120211af"
  7. )
  8. assert (
  9. build_unique_key("douyin", platform_item_id="7612592616106388138")
  10. == "dy:7612592616106388138"
  11. )
  12. def test_build_unique_key_reads_raw_payload_fallbacks():
  13. assert (
  14. build_unique_key("xiaohongshu", raw_payload={"candidate": {"source_id": "xhs-1"}})
  15. == "xhs:xhs-1"
  16. )
  17. assert (
  18. build_unique_key("douyin", raw_payload={"candidate": {"raw": {"id": "dy-1"}}})
  19. == "dy:dy-1"
  20. )
  21. def test_parse_weixin_unique_key_from_article_url():
  22. url = (
  23. "https://mp.weixin.qq.com/s?__biz=MzUxMDg4NDY1Ng=="
  24. "&mid=2247484898&idx=2&sn=5069e65919f414b96b7aec53671dc1b5"
  25. "&chksm=abc#rd"
  26. )
  27. assert (
  28. parse_weixin_unique_key(url)
  29. == "wx:MzUxMDg4NDY1Ng==:2247484898:2:5069e65919f414b96b7aec53671dc1b5"
  30. )
  31. assert build_unique_key("weixin", url=url) == parse_weixin_unique_key(url)
  32. def test_weixin_unique_key_is_none_when_url_lacks_article_parts():
  33. assert parse_weixin_unique_key("https://mp.weixin.qq.com/s/a") is None
  34. assert build_unique_key("weixin", platform_item_id="old-hash", url="https://mp.weixin.qq.com/s/a") is None