Samir Boulahtit
f20266167d
- Auto-fixed 4,496 lint issues (import sorting, modern syntax, etc.) - Added ignore rules for patterns intentional in this codebase: E402 (late imports), E712 (SQLAlchemy filters), B904 (raise from), SIM108/SIM105/SIM117 (readability preferences) - Added per-file ignores for tests and scripts - Excluded broken scripts/rename_terminology.py (has curly quotes) Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
726 lines
18 KiB
Python
726 lines
18 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Letzshop GraphQL Schema Introspection Script.
|
|
|
|
Discovers available fields on Variant, Product, and BrandUnion types
|
|
to find EAN/GTIN/barcode identifiers.
|
|
|
|
Usage:
|
|
python scripts/letzshop_introspect.py YOUR_API_KEY
|
|
"""
|
|
|
|
import json
|
|
import sys
|
|
|
|
import requests
|
|
|
|
ENDPOINT = "https://letzshop.lu/graphql"
|
|
|
|
# Introspection queries
|
|
QUERIES = {
|
|
"Variant": """
|
|
{
|
|
__type(name: "Variant") {
|
|
name
|
|
fields {
|
|
name
|
|
type {
|
|
name
|
|
kind
|
|
ofType { name kind }
|
|
}
|
|
}
|
|
}
|
|
}
|
|
""",
|
|
"Product": """
|
|
{
|
|
__type(name: "Product") {
|
|
name
|
|
fields {
|
|
name
|
|
type {
|
|
name
|
|
kind
|
|
ofType { name kind }
|
|
}
|
|
}
|
|
}
|
|
}
|
|
""",
|
|
"BrandUnion": """
|
|
{
|
|
__type(name: "BrandUnion") {
|
|
name
|
|
kind
|
|
possibleTypes {
|
|
name
|
|
}
|
|
}
|
|
}
|
|
""",
|
|
"Brand": """
|
|
{
|
|
__type(name: "Brand") {
|
|
name
|
|
fields {
|
|
name
|
|
type {
|
|
name
|
|
kind
|
|
ofType { name kind }
|
|
}
|
|
}
|
|
}
|
|
}
|
|
""",
|
|
"InventoryUnit": """
|
|
{
|
|
__type(name: "InventoryUnit") {
|
|
name
|
|
fields {
|
|
name
|
|
type {
|
|
name
|
|
kind
|
|
ofType { name kind }
|
|
}
|
|
}
|
|
}
|
|
}
|
|
""",
|
|
"TradeId": """
|
|
{
|
|
__type(name: "TradeId") {
|
|
name
|
|
kind
|
|
fields {
|
|
name
|
|
type {
|
|
name
|
|
kind
|
|
ofType { name kind }
|
|
}
|
|
}
|
|
}
|
|
}
|
|
""",
|
|
"TradeIdParser": """
|
|
{
|
|
__type(name: "TradeIdParser") {
|
|
name
|
|
kind
|
|
enumValues { name }
|
|
}
|
|
}
|
|
""",
|
|
"Order": """
|
|
{
|
|
__type(name: "Order") {
|
|
name
|
|
fields {
|
|
name
|
|
type {
|
|
name
|
|
kind
|
|
ofType { name kind }
|
|
}
|
|
}
|
|
}
|
|
}
|
|
""",
|
|
"User": """
|
|
{
|
|
__type(name: "User") {
|
|
name
|
|
fields {
|
|
name
|
|
type {
|
|
name
|
|
kind
|
|
ofType { name kind }
|
|
}
|
|
}
|
|
}
|
|
}
|
|
""",
|
|
"Address": """
|
|
{
|
|
__type(name: "Address") {
|
|
name
|
|
fields {
|
|
name
|
|
type {
|
|
name
|
|
kind
|
|
ofType { name kind }
|
|
}
|
|
}
|
|
}
|
|
}
|
|
""",
|
|
"Shipment": """
|
|
{
|
|
__type(name: "Shipment") {
|
|
name
|
|
fields {
|
|
name
|
|
type {
|
|
name
|
|
kind
|
|
ofType { name kind }
|
|
}
|
|
}
|
|
}
|
|
}
|
|
""",
|
|
"Tracking": """
|
|
{
|
|
__type(name: "Tracking") {
|
|
name
|
|
kind
|
|
fields {
|
|
name
|
|
type {
|
|
name
|
|
kind
|
|
ofType { name kind }
|
|
}
|
|
}
|
|
}
|
|
}
|
|
""",
|
|
"ShipmentState": """
|
|
{
|
|
__type(name: "ShipmentState") {
|
|
name
|
|
kind
|
|
enumValues { name description }
|
|
}
|
|
}
|
|
""",
|
|
}
|
|
|
|
|
|
def run_query(api_key: str, query: str) -> dict:
|
|
"""Execute a GraphQL query."""
|
|
headers = {
|
|
"Authorization": f"Bearer {api_key}",
|
|
"Content-Type": "application/json",
|
|
}
|
|
response = requests.post(
|
|
ENDPOINT,
|
|
json={"query": query},
|
|
headers=headers,
|
|
timeout=30,
|
|
)
|
|
return response.json()
|
|
|
|
|
|
def format_type(type_info: dict) -> str:
|
|
"""Format a GraphQL type for display."""
|
|
if not type_info:
|
|
return "?"
|
|
|
|
kind = type_info.get("kind", "")
|
|
name = type_info.get("name", "")
|
|
of_type = type_info.get("ofType")
|
|
|
|
if kind == "NON_NULL":
|
|
return f"{format_type(of_type)}!"
|
|
if kind == "LIST":
|
|
return f"[{format_type(of_type)}]"
|
|
return name or kind
|
|
|
|
|
|
def print_fields(type_data: dict, highlight_terms: list[str] = None):
|
|
"""Print fields from introspection result."""
|
|
if not type_data:
|
|
print(" (no data)")
|
|
return
|
|
|
|
highlight_terms = highlight_terms or []
|
|
fields = type_data.get("fields") or []
|
|
|
|
if not fields:
|
|
# Might be a union type
|
|
possible_types = type_data.get("possibleTypes")
|
|
if possible_types:
|
|
print(f" Union of: {', '.join(t['name'] for t in possible_types)}")
|
|
return
|
|
|
|
# Might be an enum
|
|
enum_values = type_data.get("enumValues")
|
|
if enum_values:
|
|
print(f" Enum values: {', '.join(v['name'] for v in enum_values)}")
|
|
return
|
|
|
|
return
|
|
|
|
for field in sorted(fields, key=lambda f: f["name"]):
|
|
name = field["name"]
|
|
type_str = format_type(field.get("type", {}))
|
|
|
|
# Highlight interesting fields
|
|
marker = ""
|
|
name_lower = name.lower()
|
|
if any(term in name_lower for term in highlight_terms):
|
|
marker = " <<<< LOOK!"
|
|
|
|
print(f" {name}: {type_str}{marker}")
|
|
|
|
|
|
TEST_SHIPMENT_QUERY_UNCONFIRMED = """
|
|
query {
|
|
shipments(state: unconfirmed) {
|
|
nodes {
|
|
id
|
|
number
|
|
state
|
|
order {
|
|
id
|
|
number
|
|
email
|
|
total
|
|
completedAt
|
|
locale
|
|
shipAddress {
|
|
firstName
|
|
lastName
|
|
merchant
|
|
streetName
|
|
streetNumber
|
|
city
|
|
zipCode
|
|
phone
|
|
country {
|
|
name { en fr de }
|
|
iso
|
|
}
|
|
}
|
|
billAddress {
|
|
firstName
|
|
lastName
|
|
merchant
|
|
streetName
|
|
streetNumber
|
|
city
|
|
zipCode
|
|
phone
|
|
country {
|
|
name { en fr de }
|
|
iso
|
|
}
|
|
}
|
|
}
|
|
inventoryUnits {
|
|
id
|
|
state
|
|
variant {
|
|
id
|
|
sku
|
|
mpn
|
|
price
|
|
tradeId {
|
|
number
|
|
parser
|
|
}
|
|
product {
|
|
name {
|
|
en
|
|
fr
|
|
de
|
|
}
|
|
_brand {
|
|
... on Brand {
|
|
name
|
|
}
|
|
}
|
|
}
|
|
}
|
|
}
|
|
tracking {
|
|
code
|
|
provider
|
|
}
|
|
}
|
|
}
|
|
}
|
|
"""
|
|
|
|
TEST_SHIPMENT_QUERY_CONFIRMED = """
|
|
query {
|
|
shipments(state: confirmed) {
|
|
nodes {
|
|
id
|
|
number
|
|
state
|
|
order {
|
|
id
|
|
number
|
|
email
|
|
total
|
|
completedAt
|
|
locale
|
|
shipAddress {
|
|
firstName
|
|
lastName
|
|
merchant
|
|
streetName
|
|
streetNumber
|
|
city
|
|
zipCode
|
|
phone
|
|
country {
|
|
name { en fr de }
|
|
iso
|
|
}
|
|
}
|
|
billAddress {
|
|
firstName
|
|
lastName
|
|
merchant
|
|
streetName
|
|
streetNumber
|
|
city
|
|
zipCode
|
|
phone
|
|
country {
|
|
name { en fr de }
|
|
iso
|
|
}
|
|
}
|
|
}
|
|
inventoryUnits {
|
|
id
|
|
state
|
|
variant {
|
|
id
|
|
sku
|
|
mpn
|
|
price
|
|
tradeId {
|
|
number
|
|
parser
|
|
}
|
|
product {
|
|
name {
|
|
en
|
|
fr
|
|
de
|
|
}
|
|
_brand {
|
|
... on Brand {
|
|
name
|
|
}
|
|
}
|
|
}
|
|
}
|
|
}
|
|
tracking {
|
|
code
|
|
provider
|
|
}
|
|
}
|
|
}
|
|
}
|
|
"""
|
|
|
|
|
|
TEST_SHIPMENT_QUERY_SIMPLE = """
|
|
query {{
|
|
shipments(state: {state}) {{
|
|
nodes {{
|
|
id
|
|
number
|
|
state
|
|
order {{
|
|
id
|
|
number
|
|
email
|
|
total
|
|
locale
|
|
shipAddress {{
|
|
firstName
|
|
lastName
|
|
city
|
|
zipCode
|
|
country {{
|
|
iso
|
|
}}
|
|
}}
|
|
}}
|
|
inventoryUnits {{
|
|
id
|
|
state
|
|
variant {{
|
|
id
|
|
sku
|
|
mpn
|
|
price
|
|
tradeId {{
|
|
number
|
|
parser
|
|
}}
|
|
product {{
|
|
name {{
|
|
en
|
|
fr
|
|
de
|
|
}}
|
|
}}
|
|
}}
|
|
}}
|
|
}}
|
|
}}
|
|
}}
|
|
"""
|
|
|
|
|
|
def test_shipment_query(api_key: str, state: str = "unconfirmed"):
|
|
"""Test the full shipment query with all new fields."""
|
|
print("\n" + "=" * 60)
|
|
print(f"Testing Full Shipment Query (state: {state})")
|
|
print("=" * 60)
|
|
|
|
# Try simple query first (without _brand which may cause issues)
|
|
query = TEST_SHIPMENT_QUERY_SIMPLE.format(state=state)
|
|
|
|
try:
|
|
result = run_query(api_key, query)
|
|
|
|
if "errors" in result:
|
|
print("\n❌ QUERY FAILED!")
|
|
print(f"Errors: {json.dumps(result['errors'], indent=2)}")
|
|
return False
|
|
|
|
shipments = result.get("data", {}).get("shipments", {}).get("nodes", [])
|
|
print(f"\n✅ Query successful! Found {len(shipments)} unconfirmed shipment(s)")
|
|
|
|
if shipments:
|
|
# Show first shipment as example
|
|
shipment = shipments[0]
|
|
order = shipment.get("order", {})
|
|
units = shipment.get("inventoryUnits", [])
|
|
|
|
print("\nExample shipment:")
|
|
print(f" Shipment #: {shipment.get('number')}")
|
|
print(f" Order #: {order.get('number')}")
|
|
print(f" Customer: {order.get('email')}")
|
|
print(f" Locale: {order.get('locale')} <<<< LANGUAGE")
|
|
print(f" Total: {order.get('total')} EUR")
|
|
|
|
ship_addr = order.get("shipAddress", {})
|
|
country = ship_addr.get("country", {})
|
|
print(f"\n Ship to: {ship_addr.get('firstName')} {ship_addr.get('lastName')}")
|
|
print(f" City: {ship_addr.get('zipCode')} {ship_addr.get('city')}")
|
|
print(f" Country: {country.get('iso')}")
|
|
|
|
print(f"\n Items ({len(units)}):")
|
|
for unit in units[:3]: # Show first 3
|
|
variant = unit.get("variant", {})
|
|
product = variant.get("product", {})
|
|
trade_id = variant.get("tradeId") or {}
|
|
|
|
name = product.get("name", {})
|
|
product_name = name.get("en") or name.get("fr") or name.get("de") or "?"
|
|
|
|
print(f"\n - {product_name}")
|
|
print(f" SKU: {variant.get('sku')}")
|
|
print(f" MPN: {variant.get('mpn')}")
|
|
print(f" EAN: {trade_id.get('number')} ({trade_id.get('parser')}) <<<< BARCODE")
|
|
print(f" Price: {variant.get('price')} EUR")
|
|
|
|
if len(units) > 3:
|
|
print(f"\n ... and {len(units) - 3} more items")
|
|
|
|
return True
|
|
|
|
except Exception as e:
|
|
print(f"\n❌ ERROR: {e}")
|
|
return False
|
|
|
|
|
|
def main():
|
|
if len(sys.argv) < 2:
|
|
print("Usage: python scripts/letzshop_introspect.py YOUR_API_KEY [OPTIONS]")
|
|
print("\nThis script queries the Letzshop GraphQL schema to discover")
|
|
print("available fields for EAN, GTIN, barcode, brand, etc.")
|
|
print("\nOptions:")
|
|
print(" --test Run the full shipment query to verify it works")
|
|
print(" --confirmed Test with confirmed shipments (default: unconfirmed)")
|
|
print(" --tracking Test tracking API workarounds (investigate bug)")
|
|
sys.exit(1)
|
|
|
|
api_key = sys.argv[1]
|
|
run_test = "--test" in sys.argv
|
|
use_confirmed = "--confirmed" in sys.argv
|
|
|
|
# Terms to highlight in output
|
|
highlight = ["ean", "gtin", "barcode", "brand", "mpn", "sku", "code", "identifier", "lang", "locale", "country"]
|
|
|
|
print("=" * 60)
|
|
print("Letzshop GraphQL Schema Introspection")
|
|
print("=" * 60)
|
|
print(f"\nLooking for fields containing: {', '.join(highlight)}\n")
|
|
|
|
for type_name, query in QUERIES.items():
|
|
print(f"\n{'='*60}")
|
|
print(f"Type: {type_name}")
|
|
print("=" * 60)
|
|
|
|
try:
|
|
result = run_query(api_key, query)
|
|
|
|
if "errors" in result:
|
|
print(f" ERROR: {result['errors']}")
|
|
continue
|
|
|
|
type_data = result.get("data", {}).get("__type")
|
|
if type_data:
|
|
print_fields(type_data, highlight)
|
|
else:
|
|
print(" (type not found)")
|
|
|
|
except Exception as e:
|
|
print(f" ERROR: {e}")
|
|
|
|
print("\n" + "=" * 60)
|
|
print("Done! Look for '<<<< LOOK!' markers for relevant fields.")
|
|
print("=" * 60)
|
|
|
|
# Run the test query if requested
|
|
if run_test:
|
|
state = "confirmed" if use_confirmed else "unconfirmed"
|
|
test_shipment_query(api_key, state)
|
|
|
|
# Test tracking workaround if requested
|
|
if "--tracking" in sys.argv:
|
|
test_tracking_workaround(api_key)
|
|
|
|
|
|
def test_tracking_workaround(api_key: str):
|
|
"""
|
|
Test various approaches to get tracking information.
|
|
|
|
Known issue: The `tracking` field causes a server error.
|
|
This function tests alternative approaches.
|
|
"""
|
|
print("\n" + "=" * 60)
|
|
print("Testing Tracking Workarounds")
|
|
print("=" * 60)
|
|
|
|
# Test 1: Query shipment without tracking field
|
|
print("\n1. Shipment query WITHOUT tracking field:")
|
|
query_no_tracking = """
|
|
query {
|
|
shipments(state: confirmed, first: 1) {
|
|
nodes {
|
|
id
|
|
number
|
|
state
|
|
}
|
|
}
|
|
}
|
|
"""
|
|
try:
|
|
result = run_query(api_key, query_no_tracking)
|
|
if "errors" in result:
|
|
print(f" ❌ FAILED: {result['errors']}")
|
|
else:
|
|
shipments = result.get("data", {}).get("shipments", {}).get("nodes", [])
|
|
print(f" ✅ SUCCESS: Found {len(shipments)} shipments")
|
|
if shipments:
|
|
print(f" Sample: {shipments[0]}")
|
|
except Exception as e:
|
|
print(f" ❌ ERROR: {e}")
|
|
|
|
# Test 2: Query shipment WITH tracking field (expected to fail)
|
|
print("\n2. Shipment query WITH tracking field (expected to fail):")
|
|
query_with_tracking = """
|
|
query {
|
|
shipments(state: confirmed, first: 1) {
|
|
nodes {
|
|
id
|
|
number
|
|
state
|
|
tracking {
|
|
code
|
|
provider
|
|
}
|
|
}
|
|
}
|
|
}
|
|
"""
|
|
try:
|
|
result = run_query(api_key, query_with_tracking)
|
|
if "errors" in result:
|
|
print(f" ❌ FAILED (expected): {result['errors'][0].get('message', 'Unknown error')}")
|
|
else:
|
|
print(f" ✅ SUCCESS (unexpected!): {result}")
|
|
except Exception as e:
|
|
print(f" ❌ ERROR: {e}")
|
|
|
|
# Test 3: Try to query Tracking type directly via introspection
|
|
print("\n3. Introspecting Tracking type:")
|
|
try:
|
|
result = run_query(api_key, QUERIES.get("Tracking", ""))
|
|
if "errors" in result:
|
|
print(f" ❌ FAILED: {result['errors']}")
|
|
else:
|
|
type_data = result.get("data", {}).get("__type")
|
|
if type_data:
|
|
print(" ✅ Tracking type found. Fields:")
|
|
print_fields(type_data, ["code", "provider", "carrier", "number"])
|
|
else:
|
|
print(" ⚠️ Tracking type not found in schema")
|
|
except Exception as e:
|
|
print(f" ❌ ERROR: {e}")
|
|
|
|
# Test 4: Check if there are alternative tracking-related fields on Shipment
|
|
print("\n4. Looking for alternative tracking fields on Shipment:")
|
|
try:
|
|
result = run_query(api_key, QUERIES.get("Shipment", ""))
|
|
if "errors" in result:
|
|
print(f" ❌ FAILED: {result['errors']}")
|
|
else:
|
|
type_data = result.get("data", {}).get("__type")
|
|
if type_data:
|
|
fields = type_data.get("fields", [])
|
|
tracking_related = [
|
|
f for f in fields
|
|
if any(term in f["name"].lower() for term in
|
|
["track", "carrier", "ship", "deliver", "dispatch", "fulfil"])
|
|
]
|
|
if tracking_related:
|
|
print(" Found potential tracking-related fields:")
|
|
for f in tracking_related:
|
|
print(f" - {f['name']}: {format_type(f.get('type', {}))}")
|
|
else:
|
|
print(" No alternative tracking fields found")
|
|
except Exception as e:
|
|
print(f" ❌ ERROR: {e}")
|
|
|
|
print("\n" + "=" * 60)
|
|
print("Tracking Workaround Summary:")
|
|
print("-" * 60)
|
|
print("""
|
|
Current status: Letzshop API has a bug where querying the 'tracking'
|
|
field causes a server error (NoMethodError: undefined method 'demodulize').
|
|
|
|
Workaround options:
|
|
1. Wait for Letzshop to fix the bug
|
|
2. Query shipments without tracking field, then retrieve tracking
|
|
info via a separate mechanism (e.g., Letzshop merchant portal)
|
|
3. Check if tracking info is available via webhook notifications
|
|
4. Store tracking info locally after setting it via confirmInventoryUnit
|
|
|
|
Recommendation: For now, skip querying tracking and rely on local
|
|
tracking data after confirmation via API.
|
|
""")
|
|
print("=" * 60)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|