Run DatabaseComparator on each new_item; keep only new/update items.
(self, update: dict)
| 2899 | return result |
| 2900 | |
| 2901 | def _apply_db_comparison(self, update: dict) -> Optional[dict]: |
| 2902 | """Run DatabaseComparator on each new_item; keep only new/update items.""" |
| 2903 | new_items = update.get("new_items", []) |
| 2904 | if not new_items: |
| 2905 | return update |
| 2906 | category = update.get("category", "") |
| 2907 | confirmed_items = [] |
| 2908 | noise_filtered = 0 |
| 2909 | for item in new_items: |
| 2910 | title = item.get("title", "") |
| 2911 | link = item.get("link", "") |
| 2912 | date = item.get("pub_date", "") |
| 2913 | classification, desc = self.db_comparator.classify(category, title, link, date) |
| 2914 | item["db_classification"] = classification |
| 2915 | item["db_description"] = desc |
| 2916 | if classification in ("new", "update"): |
| 2917 | confirmed_items.append(item) |
| 2918 | else: |
| 2919 | noise_filtered += 1 |
| 2920 | update["noise_filtered"] = noise_filtered |
| 2921 | update["db_confirmed_items"] = confirmed_items |
| 2922 | if not confirmed_items: |
| 2923 | print(f" INFO: All {len(new_items)} items filtered by DB comparison (noise)") |
| 2924 | return None |
| 2925 | update["new_items"] = confirmed_items |
| 2926 | update["note"] = ( |
| 2927 | f"{len(confirmed_items)} DB-confirmed new item(s) " |
| 2928 | f"({noise_filtered} noise filtered from {len(new_items)} total)" |
| 2929 | ) |
| 2930 | update["db_confirmed"] = True |
| 2931 | return update |
| 2932 | |
| 2933 | def check_regulation(self, regulation: str) -> list: |
| 2934 | if regulation not in SOURCES: |
no test coverage detected