vreau-digital/services/seap-scraper/test_seap_demo.py

#!/usr/bin/env python3
"""
Minimal SEAP WSP connectivity test.
SEAP_ENV=prod  -> https://e-licitatie.ro:8883/pub  (mTLS, requires
                  credentials/client.crt + client.key extracted from p12)
SEAP_ENV=demo  -> http://demo.e-licitatie.ro:8884/pub  (no TLS)
SEAP_OP=<name> -> SuContracts (default), CANotices, PINotices, SU_CaNotices

Extract cert+key on demand (do NOT commit):
  source ~/Code/claude-dotfiles/load-infisical-path.sh /seap
  cd credentials
  openssl pkcs12 -in 50076FB3826FADA540ACFB19.p12 -clcerts -nokeys \\
    -passin env:SEAP_CERT_KEY > client.crt
  openssl pkcs12 -in 50076FB3826FADA540ACFB19.p12 -nocerts -nodes \\
    -passin env:SEAP_CERT_KEY > client.key
  chmod 600 client.{crt,key}
"""
import os
import sys

from lxml import etree
from zeep import Client, Settings
from zeep.plugins import HistoryPlugin
from zeep.transports import Transport

DEMO_WSDL = 'http://demo.e-licitatie.ro:8884/pub?wsdl'
LOCAL_WSDL = os.path.join(os.path.dirname(__file__), 'wsdl', 'pub.wsdl')

SICAP_NS = 'http://schemas.datacontract.org/2004/07/SICAP.Service.Integration'
TEMPURI_NS = 'http://tempuri.org/'


def build_credentials_header(user: str, password: str) -> etree._Element:
    cred = etree.Element('{%s}SeapUserCredentials' % TEMPURI_NS, nsmap={
        None: TEMPURI_NS,
        'i': 'http://www.w3.org/2001/XMLSchema-instance',
    })
    pw = etree.SubElement(cred, '{%s}Password' % SICAP_NS, nsmap={None: SICAP_NS})
    pw.text = password
    un = etree.SubElement(cred, '{%s}Username' % SICAP_NS, nsmap={None: SICAP_NS})
    un.text = user
    return cred


def main():
    user = os.environ.get('SEAP_USER')
    pw = os.environ.get('SEAP_PASS')
    if not user or not pw:
        print('ERROR: SEAP_USER / SEAP_PASS not in env', file=sys.stderr)
        sys.exit(2)

    wsdl_source = LOCAL_WSDL if os.path.exists(LOCAL_WSDL) else DEMO_WSDL
    use_prod = os.environ.get('SEAP_ENV') == 'prod'
    endpoint = 'https://e-licitatie.ro:8883/pub' if use_prod else 'http://demo.e-licitatie.ro:8884/pub'
    print(f'WSDL:     {wsdl_source}')
    print(f'Endpoint: {endpoint}')

    history = HistoryPlugin()
    settings = Settings(strict=False, xml_huge_tree=True, raw_response=False)

    import requests as _req
    session = _req.Session()
    if use_prod:
        cred_dir = os.path.join(os.path.dirname(__file__), 'credentials')
        session.cert = (os.path.join(cred_dir, 'client.crt'),
                        os.path.join(cred_dir, 'client.key'))

    transport = Transport(timeout=30, session=session)
    client = Client(wsdl_source, settings=settings, transport=transport,
                    plugins=[history])

    service = client.create_service(
        '{http://tempuri.org/}IPubServiceEndpoint',
        endpoint,
    )

    cred_header = build_credentials_header(user, pw)

    operation = os.environ.get('SEAP_OP', 'SuContracts')

    requests = {
        'SuContracts': {
            'PageIndex': 1,
            'ContractStartDate': '2024-01-01T00:00:00',
            'ContractEndDate': '2026-05-05T00:00:00',
        },
        'CANotices': {
            'PageIndex': 1,
            'PublicationStartDate': '2026-04-01T00:00:00',
            'PublicationEndDate': '2026-05-05T00:00:00',
        },
        'PINotices': {
            'PageIndex': 1,
            'PublicationStartDate': '2026-04-01T00:00:00',
            'PublicationEndDate': '2026-05-05T00:00:00',
        },
        'SU_CaNotices': {
            'PageIndex': 1,
            'PublicationStartDate': '2026-04-01T00:00:00',
            'PublicationEndDate': '2026-05-05T00:00:00',
        },
    }

    print(f'\n--- Calling {operation} ---')
    try:
        result = getattr(service, operation)(
            request=requests[operation],
            _soapheaders=[cred_header],
        )
    except Exception as e:
        print(f'SOAP fault / error: {type(e).__name__}: {e}')
        if history.last_sent is not None:
            print('\n--- Last sent envelope (truncated) ---')
            sent = etree.tostring(history.last_sent['envelope'], pretty_print=True).decode()
            # redact password
            sent = sent.replace(pw, '<REDACTED>')
            print(sent[:2000])
        if history.last_received is not None:
            print('\n--- Last received envelope (truncated) ---')
            print(etree.tostring(history.last_received['envelope'],
                                 pretty_print=True).decode()[:2000])
        sys.exit(1)

    if history.last_sent is not None:
        sent = etree.tostring(history.last_sent['envelope'], pretty_print=True).decode()
        sent = sent.replace(pw, '<REDACTED>')
        print('\n--- Sent envelope ---')
        print(sent)
    if history.last_received is not None:
        recv = etree.tostring(history.last_received['envelope'], pretty_print=True).decode()
        print('\n--- Received envelope (first 3000 chars) ---')
        print(recv[:3000])
        print('\n--- Result ---')

    print(f'Status:      {getattr(result, "Status", "?")}')
    print(f'Description: {getattr(result, "Description", None)}')
    print(f'PageIndex:   {getattr(result, "PageIndex", "?")}')
    print(f'PageTotal:   {getattr(result, "PageTotal", "?")}')
    items = getattr(result, 'Items', None)
    if items is not None:
        for child_field in ('ContractItem', 'CANoticeItem', 'PINoticeItem',
                            'NoticeItem', 'Item'):
            sub = getattr(items, child_field, None)
            if sub:
                print(f'Items count ({child_field}): {len(sub)}')
                print(f'First item fields: {list(sub[0].__values__.keys())[:15]}')
                break
        else:
            print('Items present but unknown shape')


if __name__ == '__main__':
    main()