Method build_inventory

scripts/source_inventory.py:212–281

(
        self,
        current_sources: Optional[List[Dict]] = None,
        candidate_sources: Optional[List[Dict]] = None,
        *,
        save: bool = True,
    )

        return boosted

    def build_inventory(
        self,
        current_sources: Optional[List[Dict]] = None,
        candidate_sources: Optional[List[Dict]] = None,
        *,
        save: bool = True,
    ) -> Tuple[List[Dict], List[Dict], Dict]:
        current = deepcopy(current_sources if current_sources is not None else self.load_working_sources())
        candidates = deepcopy(candidate_sources if candidate_sources is not None else self.load_candidate_sources())

        screened_current, _, _ = self.policy.screen_sources(current)
        screened_candidates, _ = self.refresh_candidate_pool(candidates, save=False)

        merged = self._dedupe_by_url(screened_current + screened_candidates)
        boosted = self._apply_existing_bonus(screened_current, merged)

        working_selector = SourceSelector(
            max_per_domain=self.max_per_domain,
            target_count=min(self.working_target, self.max_working_sources),
        )
        working_sources, working_stats = working_selector.select(boosted, strategy="domain_diversity")

        export_selector = SourceSelector(
            max_per_domain=self.max_per_domain,
            target_count=self.export_target,
        )
        export_sources, export_stats = export_selector.select(working_sources, strategy="domain_diversity")

        report = {
            "timestamp": datetime.now().isoformat(),
            "current_input": len(current),
            "candidate_input": len(candidates),
            "working_count": len(working_sources),
            "export_count": len(export_sources),
            "working_stats": working_stats,
            "export_stats": export_stats,
            "needs_replenishment": len(working_sources) < self.min_working_sources,
        }

        if save:
            working_temp: Optional[Path] = None
            metadata_temp: Optional[Path] = None
            try:
                working_temp = self._stage_json(self.working_file, working_sources)

                export_success = self.updater.safe_update(
                    export_sources,
                    skip_validation=len(export_sources) < self.updater.MIN_SOURCES
                )
                if not export_success:
                    raise RuntimeError("导出书源更新失败")

                self._commit_staged_json(working_temp, self.working_file)
                working_temp = None

                stats_sync = self.sync_public_stats()
                report["stats_sync"] = stats_sync
                metadata_payload = self.prepare_metadata(report, stats_sync)