feat(ingest): country-focused seeding (collect domestic CN products)
Add a market-focused seeding path so the catalogue can be built from domestic products rather than the English-heavy global default: - adapter.fetch_by_country(country): OFF search filtered by countries_tags_en, sorted by unique_scans_n (most-scanned first), de-duplicated across pages since OFF popularity ordering is unstable. - is_cn_gs1(code): True for GS1-China company prefixes (690-699), i.e. genuinely domestic items vs. imports merely sold in China. - seed_off --country <slug> [--domestic-only] [--page-size/--max-pages]: e.g. 'seed_off --country china --domestic-only' loads only 69x barcodes. Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>
This commit is contained in:
@@ -0,0 +1,57 @@
|
||||
import httpx
|
||||
|
||||
from opengoods.adapters.openfoodfacts import OpenFoodFactsAdapter, is_cn_gs1
|
||||
|
||||
|
||||
def _adapter(handler, **kwargs):
|
||||
client = httpx.Client(transport=httpx.MockTransport(handler))
|
||||
return OpenFoodFactsAdapter(client=client, min_interval=0, backoff_base=0, **kwargs)
|
||||
|
||||
|
||||
def test_is_cn_gs1():
|
||||
assert is_cn_gs1("6901234567892") # GS1 China prefix
|
||||
assert is_cn_gs1("690")
|
||||
assert not is_cn_gs1("3017624010701") # France
|
||||
assert not is_cn_gs1("5449000000996") # Belgium
|
||||
assert not is_cn_gs1("")
|
||||
assert not is_cn_gs1(None)
|
||||
assert not is_cn_gs1("69") # too short to carry a prefix digit
|
||||
|
||||
|
||||
def test_fetch_by_country_passes_filter_and_paginates():
|
||||
seen_params = []
|
||||
|
||||
def handler(request: httpx.Request) -> httpx.Response:
|
||||
seen_params.append(dict(request.url.params))
|
||||
page = int(request.url.params.get("page", "1"))
|
||||
if page == 1:
|
||||
return httpx.Response(
|
||||
200,
|
||||
json={"products": [{"code": "6901"}, {"code": "6902"}]},
|
||||
)
|
||||
return httpx.Response(200, json={"products": []})
|
||||
|
||||
adapter = _adapter(handler)
|
||||
out = list(adapter.fetch_by_country("china", page_size=2, max_pages=5))
|
||||
assert [p["code"] for p in out] == ["6901", "6902"]
|
||||
assert seen_params[0]["countries_tags_en"] == "china"
|
||||
assert seen_params[0]["sort_by"] == "unique_scans_n"
|
||||
|
||||
|
||||
def test_fetch_by_country_dedupes_across_pages():
|
||||
def handler(request: httpx.Request) -> httpx.Response:
|
||||
page = int(request.url.params.get("page", "1"))
|
||||
if page == 1:
|
||||
return httpx.Response(
|
||||
200, json={"products": [{"code": "6901"}, {"code": "6902"}]}
|
||||
)
|
||||
if page == 2:
|
||||
# OFF popularity ordering is unstable; a repeat appears on page 2
|
||||
return httpx.Response(
|
||||
200, json={"products": [{"code": "6902"}, {"code": "6903"}]}
|
||||
)
|
||||
return httpx.Response(200, json={"products": []})
|
||||
|
||||
adapter = _adapter(handler)
|
||||
out = [p["code"] for p in adapter.fetch_by_country("china", page_size=2, max_pages=5)]
|
||||
assert out == ["6901", "6902", "6903"]
|
||||
Reference in New Issue
Block a user