cadrumo.application.ledger.actions_classification module

Bulk and rule-based ledger classification services.

CSV bulk classification parses BulkClassifyRow records, converts populated fields into ManualLedgerTransactionPatch, and applies them through update_manual_transaction_fields(). Rule application evaluates LedgerClassificationRule instances over active transactions and returns ApplyRulesResult.

Batch persistence uses the application co-commit ports so the end-of-batch save can compose the catalogue write with bucket events in one secure-object unit of work.

bulk_classify_from_csv(*, bucket_id, csv_text, actor, source_command='aeat app ledger classify --file', ports)[source]

Apply batch classifications from a CSV string.

The CSV must contain transaction_id and classification columns; category_id, business_pct, usage_ratio_id, taxable_base, iva_rate, iva_amount, iva_category, and irpf_category are optional. Blank optional cells are treated as omitted, so a partial CSV classification row preserves existing tax facts instead of clearing them accidentally. Populated tax facts ride the same ManualLedgerTransactionPatch and the update_manual_transaction_fields() write path the single-classify surface uses, so a bulk row persists the same typed taxable_base/iva_rate/iva_amount/iva_category/ irpf_category values as --id-mode classify with identical validation. Unknown columns are rejected before any writes. Rows that fail validation (unknown transaction id, invalid classification value, malformed tax fact, pydantic error) are collected in failures and the remaining valid rows are applied (partial-success semantics matching the ledger import pattern).

Returns a BulkClassifyResult.

Return type:

BulkClassifyResult

Parameters:
add_classification_rule(*, bucket_id, description_pattern, classification, category_id=None, priority=100, actor, rule_repository=None)[source]

Persist a new ledger classification rule and return it.

rule_id is content-addressed: adding the same description_pattern + classification + category_id combination twice produces the same id and the repository save overwrites the prior entry (idempotent creation).

Returns a LedgerClassificationRule.

Raises ValueError when description_pattern is not a valid regex, as validated by LedgerClassificationRule.

Return type:

LedgerClassificationRule

Parameters:
class ClassificationRulePlanRow(transaction_id, description, matched_rule_id, classification, category_id)[source]

Bases: NamedTuple

One transaction the stored rules would classify, and the rule that wins.

Parameters:
transaction_id: str

Alias for field number 0

description: str

Alias for field number 1

matched_rule_id: str

Alias for field number 2

classification: BusinessClassification

Alias for field number 3

category_id: str | None

Alias for field number 4

class ClassificationRulePlan(rules_evaluated, transactions_scanned, skipped_already_classified, no_match, matches)[source]

Bases: NamedTuple

What applying the stored rules right now would do, without doing it.

Carries the same counters the applied result reports, so a preview and the run it previews describe the same scan rather than two different summaries.

Parameters:
rules_evaluated: int

Alias for field number 0

transactions_scanned: int

Alias for field number 1

skipped_already_classified: int

Alias for field number 2

no_match: int

Alias for field number 3

matches: tuple[ClassificationRulePlanRow, ...]

Alias for field number 4

plan_classification_rules(*, bucket_id, reaffirm=False, ports, rule_repository=None)[source]

Resolve which stored rule would classify each in-scope transaction.

This is the ONE place scope and first-match are decided. apply writes the plan it returns and a preview renders it, so a preview cannot promise an outcome the run does not produce – the failure mode of two engines that agree only until one of them is edited.

Scope is ACTIVE transactions in NOT_YET_PROCESSED; reaffirm widens it to ACTIVE rows already classified manually. Rules are evaluated in stored priority order and the first match wins.

Parameters:
Return type:

ClassificationRulePlan

Returns:

The planned matches and the counters describing the scan.

apply_classification_rules(*, bucket_id, reaffirm=False, actor, source_command='aeat app ledger rule apply', ports, rule_repository=None)[source]

Apply stored classification rules to unclassified ACTIVE transactions.

Scope: ACTIVE transactions in NOT_YET_PROCESSED state. When reaffirm=True, also includes ACTIVE transactions where classified_by == "manual" so the operator can explicitly re-run the rule engine over manually classified rows.

Rules are evaluated in priority order (lower number = higher priority); the first matching rule wins. Match is re.search(pattern, description, re.IGNORECASE).

Returns an ApplyRulesResult.

Return type:

ApplyRulesResult

Parameters: