wangyunlong/gyxx-flow
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: expand module workflows, dynamic config, notifications and console API

Browse Source
This commit is contained in:
wangyunlong
2026-09-04 09:49:37 +08:00
parent 8df5266abb
commit b124d757b0
309 changed files with 89358 additions and 6232 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/modules/product_commerce/test_guanghe_metadata.py
+151 -27
View File
@@ -5,22 +5,97 @@ from unittest.mock import Mock, patch
from upload_video_to_guanghe import (
GuangHeUploader,
TmallTitleGenerationError,
_match_tm_item_ids_from_product_titles,
_select_video_product_names,
_tm_product_title_search_pattern,
download_videos,
)
def test_tm_product_title_search_pattern_uses_model_token() -> None:
assert _tm_product_title_search_pattern("拾影单肩包") == "%拾影%"
assert _tm_product_title_search_pattern("星云mini") == "%星云mini%"
def test_multi_style_record_prefers_style_named_by_attachment() -> None:
assert _select_video_product_names(
["阿波罗2", "宙斯3"], "宙斯3 7.30.mp4"
) == ["宙斯3"]
assert _select_video_product_names(
["阿波罗2", "宙斯3"], "7月30日.mp4"
) == ["阿波罗2", "宙斯3"]
def test_attachment_style_matching_handles_combos_and_overlapping_aliases() -> None:
assert _select_video_product_names(
["阿波罗2", "宙斯3", "星云2", "逐星"],
"宙斯3+阿波罗2+星云2+逐星.m4v",
) == ["宙斯3", "阿波罗2", "星云2", "逐星"]
assert _select_video_product_names(
["极星双肩", "极星双肩2"], "极星双肩2 新新新.mp4"
) == ["极星双肩2"]
class GuangHeMetadataTests(unittest.TestCase):
def test_formal_title_rejection_uses_non_browser_failure_type(self):
GuangHeUploader.configure_catalog_product_titles(
{"极星双肩2": ["极星双肩2双肩包轻便电脑包"]}
)
uploader = GuangHeUploader("", "", require_llm_titles=True)
try:
with patch(
"upload_video_to_guanghe.JdVideoTitleGenerator.generate",
side_effect=RuntimeError("候选不合格"),
):
with self.assertRaisesRegex(
TmallTitleGenerationError, "候选不合格"
):
uploader._generate_title(["极星双肩2"], video_key="token")
finally:
GuangHeUploader.configure_catalog_product_titles({})
def test_formal_collection_title_forbids_specific_style_names(self):
GuangHeUploader.configure_catalog_product_titles(
{
"宙斯3": ["宙斯3双肩包男士大容量"],
"阿波罗2": ["阿波罗2电脑包手提轻便"],
}
)
uploader = GuangHeUploader("", "", require_llm_titles=True)
try:
with patch(
"upload_video_to_guanghe.JdVideoTitleGenerator"
) as generator_class:
generator = generator_class.return_value
generator.generate.return_value = (
"多款通勤包袋,大容量轻便按场景切换"
)
title = uploader._generate_title(
["宙斯3", "阿波罗2"], video_key="combo-token"
)
self.assertEqual(title, "多款通勤包袋,大容量轻便按场景切换")
self.assertTrue(generator_class.call_args.kwargs["generic_collection"])
self.assertFalse(generator_class.call_args.kwargs["require_product_name"])
self.assertEqual(generator_class.call_args.kwargs["llm_timeout"], 300)
generator.generate.assert_called_once()
self.assertEqual(
generator.generate.call_args.args[0], ["宙斯3", "阿波罗2"]
)
finally:
GuangHeUploader.configure_catalog_product_titles({})
def test_downloads_with_same_attachment_name_are_isolated_by_record(self):
attachment = {"file_token": "file-token", "name": "same-name.mp4"}
with TemporaryDirectory() as tmp_dir, patch(
"upload_video_to_guanghe.DOWNLOAD_DIR", Path(tmp_dir)
), patch("upload_video_to_guanghe.run_lark") as run_lark_mock:
def fake_download(args):
output_dir = Path(args[args.index("--output") + 1])
def fake_download(args, *, cwd=None, timeout=None):
self.assertEqual(timeout, 600)
output_dir = Path(cwd)
output_dir.mkdir(parents=True, exist_ok=True)
(output_dir / attachment["name"]).write_bytes(b"video")
(output_dir / args[args.index("--output") + 1]).write_bytes(b"video")
return 0, "", ""
run_lark_mock.side_effect = fake_download
@@ -28,8 +103,9 @@ class GuangHeMetadataTests(unittest.TestCase):
first = download_videos("record-one", [attachment])
second = download_videos("record-two", [attachment])
self.assertEqual(first[0].parent.name, "record-one")
self.assertEqual(second[0].parent.name, "record-two")
self.assertEqual(first[0].parent.parent.name, "record-one")
self.assertEqual(second[0].parent.parent.name, "record-two")
self.assertEqual(first[0].parent.name, "file-token")
self.assertNotEqual(first[0], second[0])
def test_unresolved_style_uses_exact_database_product_title_match(self):
@@ -81,14 +157,20 @@ class GuangHeMetadataTests(unittest.TestCase):
def test_topic_result_card_requires_topic_and_activity_statistics(self):
self.assertTrue(
GuangHeUploader._topic_result_card_text_matches(
"# 在淘宝种草一夏 小二推荐 作品数469.5万 参与数16.2万",
"在淘宝种草一夏",
"# 我的夏日焕新清单 小二推荐 作品数469.5万 参与数16.2万",
"我的夏日焕新清单",
)
)
self.assertTrue(
GuangHeUploader._topic_result_card_text_matches(
"# 秋上新开拍了 时尚 作品数198.3万 参与数11.3万",
"秋上新",
)
)
self.assertFalse(
GuangHeUploader._topic_result_card_text_matches(
"在淘宝种草一夏",
"在淘宝种草一夏",
"我的夏日焕新清单",
"我的夏日焕新清单",
)
)
self.assertTrue(GuangHeUploader._topic_card_box_is_clickable(620, 120))
@@ -160,13 +242,12 @@ class GuangHeMetadataTests(unittest.TestCase):
self.assertGreaterEqual(len(title), 26)
self.assertLessEqual(len(title), 30)
def test_same_style_multiple_videos_get_distinct_titles_when_hermes_repeats(self):
def test_same_style_multiple_videos_get_distinct_titles_when_model_repeats(self):
uploader = object.__new__(GuangHeUploader)
repeated = "光影行星宙斯双肩包科学分区收纳轻量通勤出行"
with (
patch.object(uploader, "_search_product_features", return_value=""),
patch("upload_video_to_guanghe.HAS_HERMES", True),
patch("upload_video_to_guanghe._call_hermes", return_value=(repeated, {})),
patch("upload_video_to_guanghe.call_direct_llm", return_value=(repeated, {})),
):
first = uploader._generate_title(["宙斯"], video_index=1, video_total=2)
second = uploader._generate_title(["宙斯"], video_index=2, video_total=2)
@@ -175,13 +256,12 @@ class GuangHeMetadataTests(unittest.TestCase):
self.assertTrue(first.startswith("宙斯双肩包,"))
self.assertTrue(second.startswith("宙斯双肩包,"))
def test_same_style_repeated_hermes_copy_gets_low_similarity_bodies(self):
def test_same_style_repeated_model_copy_gets_low_similarity_bodies(self):
uploader = object.__new__(GuangHeUploader)
repeated = "光影行星宙斯双肩包,科学分区收纳,轻量通勤出行!"
with (
patch.object(uploader, "_search_product_features", return_value=""),
patch("upload_video_to_guanghe.HAS_HERMES", True),
patch("upload_video_to_guanghe._call_hermes", return_value=(repeated, {})),
patch("upload_video_to_guanghe.call_direct_llm", return_value=(repeated, {})),
):
titles = [
uploader._generate_title(["宙斯"], video_index=index, video_total=4)
@@ -202,8 +282,7 @@ class GuangHeMetadataTests(unittest.TestCase):
repeated = "科学分区收纳,轻装通勤出行"
with (
patch.object(uploader, "_search_product_features", return_value=""),
patch("upload_video_to_guanghe.HAS_HERMES", True),
patch("upload_video_to_guanghe._call_hermes", return_value=(repeated, {})),
patch("upload_video_to_guanghe.call_direct_llm", return_value=(repeated, {})),
):
first = uploader._generate_title(["宙斯"], video_index=1)
second = uploader._generate_title(["星云2"], video_index=1)
@@ -339,9 +418,8 @@ class GuangHeMetadataTests(unittest.TestCase):
uploader = object.__new__(GuangHeUploader)
with (
patch.object(uploader, "_search_product_features", return_value=""),
patch("upload_video_to_guanghe.HAS_HERMES", True),
patch(
"upload_video_to_guanghe._call_hermes",
"upload_video_to_guanghe.call_direct_llm",
return_value=("光影行星宙斯双肩包大容量笔记本电脑轻量通勤", {}),
),
):
@@ -350,7 +428,7 @@ class GuangHeMetadataTests(unittest.TestCase):
self.assertTrue(title.startswith("宙斯双肩包,"))
self.assertNotIn("包袋双肩包", title)
def test_generated_poseidon_title_cannot_be_changed_to_backpack_by_hermes(self):
def test_generated_poseidon_title_cannot_be_changed_to_backpack_by_model(self):
uploader = object.__new__(GuangHeUploader)
with (
patch.object(
@@ -358,9 +436,8 @@ class GuangHeMetadataTests(unittest.TestCase):
"_search_product_features",
return_value="双肩包、电脑包、托特包、大容量",
),
patch("upload_video_to_guanghe.HAS_HERMES", True),
patch(
"upload_video_to_guanghe._call_hermes",
"upload_video_to_guanghe.call_direct_llm",
return_value=(
"光影行星波塞冬edge双肩包暴雨通勤防泼水电脑仓从容出行",
{},
@@ -439,8 +516,8 @@ class GuangHeMetadataTests(unittest.TestCase):
with self.subTest(product_names=product_names):
uploader = object.__new__(GuangHeUploader)
calls = []
uploader._pick_content_labels = Mock(
side_effect=lambda labels: calls.append(("labels", list(labels)))
uploader._pick_content_label = Mock(
side_effect=lambda label: calls.append(("label", label))
)
uploader._pick_topic = Mock(
side_effect=lambda topic: calls.append(("topic", topic))
@@ -452,8 +529,8 @@ class GuangHeMetadataTests(unittest.TestCase):
self.assertEqual(
calls,
[
("labels", expected_labels),
("topic", "在淘宝种草一夏"),
*[("label", label) for label in expected_labels],
("topic", "我的夏日焕新清单"),
],
)
@@ -461,8 +538,55 @@ class GuangHeMetadataTests(unittest.TestCase):
self.assertEqual(GuangHeUploader.FIXED_CONTENT_LABELS, ("商品展示",))
self.assertNotIn("户外", GuangHeUploader.FIXED_CONTENT_LABELS)
def test_missing_user_selected_topic_stops_metadata_preparation(self):
uploader = object.__new__(GuangHeUploader)
uploader._pick_content_label = Mock()
uploader._pick_topic = Mock(
side_effect=RuntimeError("没有找到固定话题: #我的夏日焕新清单")
)
uploader._snapshot = Mock()
with self.assertRaisesRegex(RuntimeError, "没有找到固定话题"):
uploader._prepare_publish_metadata(["极星双肩"])
self.assertEqual(
uploader._pick_content_label.call_args_list,
[unittest.mock.call("双肩包"), unittest.mock.call("商品展示")],
)
def test_optional_content_label_failures_do_not_skip_required_topic(self):
uploader = object.__new__(GuangHeUploader)
uploader.page = Mock()
uploader._pick_content_label = Mock(
side_effect=[
RuntimeError("内容标签没有精确匹配项: #双肩包"),
RuntimeError("内容标签没有精确匹配项: #商品展示"),
]
)
uploader._pick_topic = Mock()
uploader._snapshot = Mock()
uploader._prepare_publish_metadata(["极星双肩"])
self.assertEqual(
uploader._pick_content_label.call_args_list,
[unittest.mock.call("双肩包"), unittest.mock.call("商品展示")],
)
uploader._pick_topic.assert_called_once_with("我的夏日焕新清单")
def test_metadata_uses_operator_topic_keyword(self):
uploader = object.__new__(GuangHeUploader)
calls = []
uploader._pick_content_label = Mock()
uploader._pick_topic = Mock(side_effect=lambda topic: calls.append(topic))
uploader._snapshot = Mock()
uploader._prepare_publish_metadata(["极星双肩"], topic_keyword="秋上新")
self.assertEqual(calls, ["秋上新"])
def test_fixed_topic_is_not_a_recommended_first_result(self):
self.assertEqual(GuangHeUploader.FIXED_TOPIC, "在淘宝种草一夏")
self.assertEqual(GuangHeUploader.FIXED_TOPIC, "我的夏日焕新清单")
if __name__ == "__main__":