(
tiny_tree: ET.ElementTree,
tiny_path: Path,
full_tree: ET.ElementTree,
full_path: Path,
fallback_category: str,
check_results: Dict[str, FeedCheckResult],
)
| 476 | |
| 477 | |
| 478 | def sync_tiny_to_full( |
| 479 | tiny_tree: ET.ElementTree, |
| 480 | tiny_path: Path, |
| 481 | full_tree: ET.ElementTree, |
| 482 | full_path: Path, |
| 483 | fallback_category: str, |
| 484 | check_results: Dict[str, FeedCheckResult], |
| 485 | ) -> int: |
| 486 | tiny_body = get_body(tiny_tree, tiny_path) |
| 487 | full_body = get_body(full_tree, full_path) |
| 488 | full_urls = set(collect_rss_urls(full_body)) |
| 489 | category_map = build_category_map(full_body) |
| 490 | merged_added = 0 |
| 491 | |
| 492 | for source_category, tiny_rss in collect_tiny_entries(tiny_body): |
| 493 | url = normalize_url(tiny_rss.attrib.get("xmlUrl", "")) |
| 494 | if not url: |
| 495 | continue |
| 496 | if not check_results.get(url, FeedCheckResult(False, "transient_fail", "missing_check_result")).alive: |
| 497 | continue |
| 498 | if url in full_urls: |
| 499 | continue |
| 500 | |
| 501 | target_name = source_category if source_category in category_map else fallback_category |
| 502 | target_category = ensure_category(full_body, category_map, target_name) |
| 503 | copied = copy.deepcopy(tiny_rss) |
| 504 | copied.attrib["xmlUrl"] = url |
| 505 | if copied.tail is None: |
| 506 | copied.tail = "\n" |
| 507 | if target_category.text is None: |
| 508 | target_category.text = "\n" |
| 509 | target_category.append(copied) |
| 510 | full_urls.add(url) |
| 511 | merged_added += 1 |
| 512 | |
| 513 | return merged_added |
| 514 | |
| 515 | |
| 516 | def serialize_tree(tree: ET.ElementTree) -> bytes: |
no test coverage detected