Claude VM
a6c03a091e
Moved from gov-agreg/src/pages/achizitii/* to root (drop prefix). - 22 pages migrated, 127 files total - All internal links: /achizitii/X → /X (176 occurrences fixed) - AchizitiiLayout subnav rewritten: /X paths, top-right link to vreaudigital.ro hub - BaseLayout new (vreau.digital branding, OG tags, site URL) - astro.config.mjs: site https://vreau.digital, server output (was static) - docker-compose: port 5096 (vreaudigital is 5095), container vreau-digital - deploy.sh: paths /opt/vreau-digital, log /var/log/vreau-digital-deploy.log Backend shared with gov-agreg: - PostgreSQL satra (same schemas: seap, firms, anaf, anre, ...) - Photon, Martin tiles - Infisical /vreaudigital path (DATABASE_URL etc. shared) build: PASS (npx astro check 0 errors, npm run build 5s vite + 10s server)
288 lines
12 KiB
Python
288 lines
12 KiB
Python
"""Parse SU_CaNotices items (Anunțuri de atribuire) → seap.announcements row dict."""
|
|
from __future__ import annotations
|
|
|
|
from ..xml_utils import (
|
|
find_child_local, find_local, find_path,
|
|
text_under, text_direct, int_under, decimal_under, bool_under,
|
|
datetime_under, sysitem_name, sysitem_id, to_jsonable,
|
|
)
|
|
|
|
SOURCE = 'wsp_canotice'
|
|
|
|
|
|
def parse(el) -> dict | None:
|
|
"""Extract a CaNoticeBase element (V1 or V2) into our normalized dict."""
|
|
notice_no = text_under(el, 'NoticeNo')
|
|
notice_id = int_under(el, 'CaNoticeId')
|
|
if not notice_no:
|
|
return None
|
|
|
|
general = find_child_local(el, 'General')
|
|
section1 = find_child_local(el, 'Section1')
|
|
section2 = find_child_local(el, 'Section2')
|
|
section4 = find_child_local(el, 'Section4')
|
|
section5 = find_child_local(el, 'Section5')
|
|
|
|
# Authority block: Section1_1 → CaAddresses → CaEntityInformation → EntityInformation
|
|
auth_entity = find_path(section1, 'Section1_1', 'CaAddresses')
|
|
auth_info = None
|
|
if auth_entity is not None:
|
|
auth_info = find_local(auth_entity, 'EntityInformation')
|
|
|
|
authority_name = text_direct(auth_info, 'Name') if auth_info is not None else None
|
|
authority_cui = text_direct(auth_info, 'Cif') if auth_info is not None else None
|
|
authority_address = text_direct(auth_info, 'Address') if auth_info is not None else None
|
|
authority_email = text_direct(auth_info, 'Email') if auth_info is not None else None
|
|
authority_phone = text_direct(auth_info, 'Phone') if auth_info is not None else None
|
|
authority_url = text_direct(auth_info, 'Url') if auth_info is not None else None
|
|
|
|
# NUTS code = county-ish (RO213 etc.)
|
|
nuts_el = find_local(auth_info, 'NutsCode') if auth_info is not None else None
|
|
county_code = sysitem_name(auth_info, 'NutsCode') if auth_info is not None else None
|
|
|
|
# Authority type + main activity
|
|
s1_4 = find_child_local(section1, 'Section1_4') if section1 is not None else None
|
|
authority_type = sysitem_name(s1_4, 'ContractingAuthorityType')
|
|
s1_5 = find_child_local(section1, 'Section1_5') if section1 is not None else None
|
|
main_activity = sysitem_name(s1_5, 'MainActivity')
|
|
|
|
# Section 2: contract details
|
|
s2_1 = find_child_local(section2, 'Section2_1') if section2 is not None else None
|
|
contract_title = text_under(general, 'ContractTitle') or text_under(s2_1, 'ContractName')
|
|
short_desc = text_under(s2_1, 'ShortContractDescription')
|
|
main_cpv_id = sysitem_id(s2_1, 'MainCPV')
|
|
main_cpv_code = sysitem_name(s2_1, 'MainCPV')
|
|
contract_type = sysitem_name(s2_1, 'SysAcquisitionContractType')
|
|
currency = sysitem_name(s2_1, 'Currency')
|
|
total_value = decimal_under(s2_1, 'TotalValue')
|
|
highest_offer = decimal_under(s2_1, 'HighestOffer')
|
|
lowest_offer = decimal_under(s2_1, 'LowestOffer')
|
|
has_lots = bool_under(s2_1, 'ContractHasLots')
|
|
reference_number = text_under(s2_1, 'ReferenceNumber')
|
|
|
|
# Lots — Section 5 (ContractLotList) for awarded notices, Section 2_2 otherwise
|
|
lots = _extract_lots(section5) or _extract_lots(section2)
|
|
lots_count = len(lots) if lots else None
|
|
|
|
# Award criteria — extracted from lot info or top-level
|
|
award_criteria = _extract_award_criteria(el)
|
|
|
|
# Section 4: procedure details
|
|
s4_1 = find_child_local(section4, 'Section4_1') if section4 is not None else None
|
|
procedure_type = sysitem_name(s4_1, 'SysProcedureType')
|
|
framework_agreement = bool_under(s4_1, 'FrameworkAgreement')
|
|
|
|
# Dates
|
|
publication_date = datetime_under(general, 'PublishDate')
|
|
legislation = sysitem_name(general, 'SysLegislationType')
|
|
notice_state = sysitem_name(general, 'SysNoticeState')
|
|
notice_state_id = sysitem_id(general, 'SysNoticeState')
|
|
is_utility = bool_under(general, 'IsUtility')
|
|
notice_no_joue = text_under(general, 'NoticeNoJoue')
|
|
entity_id = int_under(general, 'EntityId')
|
|
|
|
# Winners (suppliers) — extracted from lot info
|
|
winners = _extract_winners(section5)
|
|
# Pick first winner as primary; full list in lots JSONB
|
|
primary_winner = winners[0] if winners else {}
|
|
|
|
# Documents — DfNoticeFiles, CompanyFiles
|
|
documents = _extract_documents(general)
|
|
|
|
# Final award value
|
|
awarded_value = total_value
|
|
if not awarded_value and lots:
|
|
# sum lot values
|
|
try:
|
|
awarded_value = sum(
|
|
(lot.get('total_value') or 0) for lot in lots if lot.get('total_value')
|
|
) or None
|
|
except Exception:
|
|
pass
|
|
|
|
return {
|
|
# Standard columns
|
|
'type': 'ca_notice',
|
|
'ref_number': f'WSP-{notice_no}',
|
|
'authority_name': authority_name,
|
|
'authority_cui': authority_cui,
|
|
'authority_address': authority_address,
|
|
'authority_email': authority_email,
|
|
'authority_phone': authority_phone,
|
|
'authority_url': authority_url,
|
|
'authority_type': authority_type,
|
|
'authority_main_activity': main_activity,
|
|
'authority_entity_id': entity_id,
|
|
'title': contract_title[:1000] if contract_title else None,
|
|
'cpv_code': main_cpv_code,
|
|
'contract_type': contract_type,
|
|
'publication_date': publication_date,
|
|
'estimated_value': None,
|
|
'awarded_value': awarded_value,
|
|
'currency': currency,
|
|
'supplier_name': primary_winner.get('name'),
|
|
'supplier_cui': primary_winner.get('cif'),
|
|
'supplier_address': primary_winner.get('address'),
|
|
'supplier_is_sme': primary_winner.get('is_sme'),
|
|
'procedure_type': procedure_type,
|
|
'procedure_state': notice_state,
|
|
'legislation': legislation,
|
|
'lot_number': None,
|
|
'has_lots': 'da' if has_lots else 'nu' if has_lots is False else None,
|
|
'contract_has_lots': has_lots,
|
|
'lots_count': lots_count,
|
|
'joue': notice_no_joue,
|
|
'county_code': county_code,
|
|
'notice_state': notice_state,
|
|
'notice_state_id': notice_state_id,
|
|
'framework_agreement': framework_agreement,
|
|
'notice_id_internal': notice_id,
|
|
'seap_url': f'https://e-licitatie.ro/pub/notices/contract-award-notice/{notice_id}' if notice_id else None,
|
|
|
|
# Rich JSONB
|
|
'documents': documents or None,
|
|
'award_criteria': award_criteria or None,
|
|
'lots': lots or None,
|
|
'details': {
|
|
'short_description': short_desc,
|
|
'reference_number': reference_number,
|
|
'main_cpv_id': main_cpv_id,
|
|
'highest_offer': str(highest_offer) if highest_offer else None,
|
|
'lowest_offer': str(lowest_offer) if lowest_offer else None,
|
|
'is_utility': is_utility,
|
|
'all_winners': winners or None,
|
|
},
|
|
'source': SOURCE,
|
|
}
|
|
|
|
|
|
def _extract_lots(section) -> list[dict]:
|
|
"""Extract lot info — handles ContractLotList (Section5) or Lots (Section2_2)."""
|
|
if section is None:
|
|
return []
|
|
lot_list = find_local(section, 'ContractLotList')
|
|
if lot_list is None:
|
|
lot_list = find_local(section, 'Lots')
|
|
if lot_list is None:
|
|
return []
|
|
out = []
|
|
for lot in lot_list:
|
|
if etree.QName(lot.tag).localname not in ('ContractLotInfo', 'LotInfo'):
|
|
continue
|
|
lot_data = {
|
|
'lot_id': int_under(lot, 'LotID') or int_under(lot, 'ContractNo'),
|
|
'lot_no': int_under(lot, 'LotNo') or int_under(lot, 'ContractNo'),
|
|
'title': text_under(lot, 'Title') or text_under(lot, 'ContractObjectName'),
|
|
'description': text_under(lot, 'DescriptionOfProcurement') or text_under(lot, 'ContractObjectDescription'),
|
|
'cpv_code': sysitem_name(lot, 'MainCPVCode'),
|
|
'estimated_value': _decimal_or_none(text_under(lot, 'EstimatedValue')),
|
|
'total_value': _decimal_or_none(text_under(lot, 'TotalValue')),
|
|
'duration_months': int_under(lot, 'DurationInMonths'),
|
|
'duration_days': int_under(lot, 'DurationInDays'),
|
|
'currency': sysitem_name(lot, 'Currency'),
|
|
'place_of_performance': text_under(lot, 'MainSiteOrPlaceOfPerformance'),
|
|
'is_community_financed': bool_under(lot, 'IsCommunityFinanced'),
|
|
'has_options': bool_under(lot, 'HasOptions'),
|
|
'awarded_to_group': bool_under(lot, 'AwardedToGroupOfEcoOp'),
|
|
}
|
|
# Convert decimals to str for JSON safety
|
|
for k in ('estimated_value', 'total_value'):
|
|
if lot_data[k] is not None:
|
|
lot_data[k] = str(lot_data[k])
|
|
# only keep non-empty
|
|
out.append({k: v for k, v in lot_data.items() if v is not None})
|
|
return out
|
|
|
|
|
|
def _extract_winners(section5) -> list[dict]:
|
|
"""Extract winner info from ContractLotList → ContractorAddressList."""
|
|
if section5 is None:
|
|
return []
|
|
out = []
|
|
seen_cifs = set()
|
|
lot_list = find_local(section5, 'ContractLotList')
|
|
if lot_list is None:
|
|
return []
|
|
for lot in lot_list:
|
|
addr_list = find_local(lot, 'ContractorAddressList')
|
|
if addr_list is None:
|
|
continue
|
|
for sect in addr_list:
|
|
entity = find_local(sect, 'EntityInformation')
|
|
if entity is None:
|
|
continue
|
|
cif = text_direct(entity, 'Cif')
|
|
if cif in seen_cifs:
|
|
continue
|
|
if cif:
|
|
seen_cifs.add(cif)
|
|
out.append({
|
|
'name': text_direct(entity, 'Name'),
|
|
'cif': cif,
|
|
'address': text_direct(entity, 'Address'),
|
|
'email': text_direct(entity, 'Email'),
|
|
'phone': text_direct(entity, 'Phone'),
|
|
'url': text_direct(entity, 'Url'),
|
|
'nuts_code': sysitem_name(entity, 'NutsCode'),
|
|
'is_sme': bool_under(sect, 'IsSme'),
|
|
})
|
|
return out
|
|
|
|
|
|
def _extract_award_criteria(el) -> list[dict]:
|
|
"""Extract award criteria from lot info (MyPriceAwardCriterias / MyQualityAwardCriterias)."""
|
|
out = []
|
|
for crit_list_name in ('MyPriceAwardCriterias', 'MyQualityAwardCriterias'):
|
|
for crit_list in el.iter():
|
|
if etree.QName(crit_list.tag).localname != crit_list_name:
|
|
continue
|
|
for crit in crit_list:
|
|
name = text_under(crit, 'Name') or text_under(crit, 'Description')
|
|
weight = decimal_under(crit, 'Weight') or decimal_under(crit, 'Pondere')
|
|
if name or weight:
|
|
item = {'type': 'price' if 'Price' in crit_list_name else 'quality',
|
|
'name': name}
|
|
if weight is not None:
|
|
item['weight'] = str(weight)
|
|
out.append(item)
|
|
return out
|
|
|
|
|
|
def _extract_documents(general) -> list[dict]:
|
|
"""Extract document file references from DfNoticeFiles + CompanyFiles."""
|
|
if general is None:
|
|
return []
|
|
out = []
|
|
for fld in ('DfNoticeFiles', 'CompanyFiles', 'NoticeFiles'):
|
|
container = find_local(general, fld)
|
|
if container is None:
|
|
continue
|
|
for kvp in container:
|
|
key = find_local(kvp, 'key')
|
|
if key is None:
|
|
key = find_local(kvp, 'Key')
|
|
val = find_local(kvp, 'value')
|
|
if val is None:
|
|
val = find_local(kvp, 'Value')
|
|
if key is not None and val is not None:
|
|
out.append({
|
|
'type': fld,
|
|
'name': key.text,
|
|
'id': val.text,
|
|
})
|
|
return out
|
|
|
|
|
|
def _decimal_or_none(s):
|
|
if not s:
|
|
return None
|
|
try:
|
|
from decimal import Decimal
|
|
return Decimal(s)
|
|
except Exception:
|
|
return None
|
|
|
|
|
|
# late import to avoid circular
|
|
from lxml import etree # noqa: E402
|