Add national municipal financial statement API

app/jobs/warehouse/extract_municipal_financial_statements_job.rb

@@ -0,0 +1,90 @@
+class Warehouse::ExtractMunicipalFinancialStatementsJob < ApplicationJob
+  include ActiveJob::Continuable
+
+  queue_as :default
+  self.resume_errors_after_advancing = false
+
+  CIRCUIT_WINDOW = 20
+  MAX_FAILURE_RATE = 0.8
+  MAX_CONSECUTIVE_MISSING_ASSETS = 3
+  FAILURE_STATUSES = %w[failed missing_asset].freeze
+
+  class CircuitOpen < StandardError; end
+
+  def perform(release_version, province:, years: nil, institution_ids: nil,
+    limit: nil, rerun: "missing", asset_root: nil)
+    @release = Warehouse::InstitutionRelease.find_by!(version: release_version)
+    @candidate_set = candidate_set(
+      province:, years:, institution_ids:, asset_root:
+    )
+    unless @candidate_set.asset_root.directory?
+      raise CircuitOpen, "asset root is unavailable: #{@candidate_set.asset_root}"
+    end
+    @processor = processor(rerun:)
+    @limit = Integer(limit) if limit
+    @attempted = 0
+    @recent_outcomes = []
+    @consecutive_missing_assets = 0
+
+    step :extract_statements do |step|
+      @candidate_set.each(start: step.cursor) do |candidate|
+        result = @processor.call(candidate)
+        log_result(candidate, result)
+        track_result!(result)
+        @attempted += 1 unless result.status == "skipped"
+        step.advance! from: candidate.document_id
+        check_circuit!
+        break if @limit && @attempted >= @limit
+      end
+    end
+  end
+
+  private
+
+  def candidate_set(province:, years:, institution_ids:, asset_root:)
+    options = {
+      release: @release, provinces: [ province ], years:, institution_ids:
+    }
+    options[:asset_root] = asset_root if asset_root.present?
+    Warehouse::FinancialStatementExtraction::CandidateSet.new(**options)
+  end
+
+  def processor(rerun:)
+    Warehouse::FinancialStatementExtraction::Processor.new(release: @release, rerun:)
+  end
+
+  def track_result!(result)
+    @recent_outcomes << result.status
+    @recent_outcomes.shift while @recent_outcomes.length > CIRCUIT_WINDOW
+    @consecutive_missing_assets = if result.status == "missing_asset"
+      @consecutive_missing_assets + 1
+    else
+      0
+    end
+  end
+
+  def check_circuit!
+    if @consecutive_missing_assets >= MAX_CONSECUTIVE_MISSING_ASSETS
+      raise CircuitOpen, "#{@consecutive_missing_assets} consecutive archived assets are missing"
+    end
+    return unless @recent_outcomes.length == CIRCUIT_WINDOW
+
+    failures = @recent_outcomes.count { _1.in?(FAILURE_STATUSES) }
+    if failures.fdiv(CIRCUIT_WINDOW) >= MAX_FAILURE_RATE
+      raise CircuitOpen, "#{failures} of the last #{CIRCUIT_WINDOW} statements failed"
+    end
+  end
+
+  def log_result(candidate, result)
+    payload = {
+      event: "municipal_financial_statement_extraction",
+      document_id: candidate.document_canonical_id,
+      fiscal_year: candidate.fiscal_year_end.year,
+      status: result.status,
+      stage: result.stage,
+      extraction_id: result.extraction_id,
+      error: result.error
+    }.compact
+    Rails.logger.info(payload.to_json)
+  end
+end

app/models/warehouse/census_profile.rb

@@ -0,0 +1,12 @@
+class Warehouse::CensusProfile < Warehouse::Record
+  GEO_LEVELS = %w[csd cd pr da].freeze
+
+  validates :census_year, numericality: { only_integer: true, greater_than: 0 }
+  validates :geo_level, inclusion: { in: GEO_LEVELS }
+  validates :geo_uid, :source_url, :retrieved_at, presence: true
+  validates :population, numericality: { only_integer: true, greater_than: 0 }
+  validates :area_sq_km, numericality: { greater_than: 0 }, allow_nil: true
+  validates :population_density_per_sq_km, numericality: { greater_than_or_equal_to: 0 }, allow_nil: true
+  validates :source_sha256, format: { with: /\A[0-9a-f]{64}\z/ },
+    uniqueness: { scope: [ :census_year, :geo_level, :geo_uid ] }
+end

app/models/warehouse/census_profile_importer.rb

@@ -0,0 +1,92 @@
+require "csv"
+require "digest"
+require "open3"
+
+class Warehouse::CensusProfileImporter
+  SOURCE_URL = "https://www12.statcan.gc.ca/census-recensement/2021/dp-pd/prof/details/download-telecharger/comp/getFile.cfm?LANG=E&GEONO=005&FILETYPE=CSV"
+  CSV_ENTRY = "98-401-X2021005_English_CSV_data.csv"
+
+  class ImportError < StandardError; end
+
+  def initialize(zip_path:, expected_sha256:, retrieved_at:)
+    @zip_path = Pathname(zip_path).expand_path
+    @expected_sha256 = expected_sha256
+    @retrieved_at = retrieved_at.to_time
+  end
+
+  def import!
+    validate_source!
+    profiles = read_profiles
+    now = Time.current
+    rows = profiles.map do |uid, values|
+      {
+        census_year: 2021, geo_level: "csd", geo_uid: uid,
+        population: values.fetch(:population), area_sq_km: values[:area_sq_km],
+        population_density_per_sq_km: values[:population_density_per_sq_km],
+        source_url: SOURCE_URL, source_sha256: @expected_sha256,
+        retrieved_at: @retrieved_at, created_at: now, updated_at: now
+      }
+    end
+    Warehouse::CensusProfile.insert_all(
+      rows, unique_by: "index_census_profiles_vintage_geography_source"
+    )
+    {
+      source_url: SOURCE_URL,
+      source_sha256: @expected_sha256,
+      census_profile_rows: profiles.length,
+      stored_profiles: Warehouse::CensusProfile.where(
+        census_year: 2021, geo_level: "csd", source_sha256: @expected_sha256
+      ).count
+    }
+  end
+
+  private
+
+  def validate_source!
+    raise ImportError, "missing Census Profile archive #{@zip_path}" unless @zip_path.file?
+    actual = Digest::SHA256.file(@zip_path).hexdigest
+    raise ImportError, "Census Profile SHA-256 mismatch: expected #{@expected_sha256}, got #{actual}" unless actual == @expected_sha256
+  end
+
+  def read_profiles
+    rows = Hash.new { |hash, uid| hash[uid] = {} }
+    Open3.popen3("unzip", "-p", @zip_path.to_s, CSV_ENTRY) do |stdin, stdout, stderr, wait|
+      stdin.close
+      stdout.set_encoding(Encoding::Windows_1252, Encoding::UTF_8, invalid: :replace, undef: :replace)
+      headers = CSV.parse_line(stdout.gets)
+      indexes = %w[DGUID GEO_LEVEL CHARACTERISTIC_ID C1_COUNT_TOTAL]
+        .to_h { |name| [ name, headers.index(name) || raise(ImportError, "missing Census Profile column #{name}") ] }
+      stdout.each_line do |line|
+        next unless line.include?("Census subdivision")
+        next unless line.match?(/,(?:1|6|7),"(?:Population, 2021|Population density per square kilometre|Land area in square kilometres)",/)
+
+        values = CSV.parse_line(line)
+        next unless values[indexes.fetch("GEO_LEVEL")] == "Census subdivision"
+
+        uid = values[indexes.fetch("DGUID")].to_s.delete_prefix("2021A0005")
+        next unless uid.length == 7
+
+        value = values[indexes.fetch("C1_COUNT_TOTAL")]
+        case values[indexes.fetch("CHARACTERISTIC_ID")]
+        when "1"
+          population = Integer(value, exception: false)
+          rows[uid][:population] = population if population&.positive?
+        when "6"
+          density = BigDecimal(value, exception: false)
+          rows[uid][:population_density_per_sq_km] = density if density && density >= 0
+        when "7"
+          area = BigDecimal(value, exception: false)
+          rows[uid][:area_sq_km] = area if area&.positive?
+        end
+      end
+      error = stderr.read
+      raise ImportError, "unzip failed: #{error}" unless wait.value.success?
+    end
+    rows.select! { |_uid, values| values[:population]&.positive? }
+    raise ImportError, "Census Profile contained no CSD profile rows" if rows.empty?
+
+    rows
+  rescue CSV::MalformedCSVError => error
+    raise ImportError, "invalid Census Profile CSV: #{error.message}"
+  end
+end

app/models/warehouse/financial_statement_extraction.rb

@@ -10,24 +10,44 @@ class Warehouse::FinancialStatementExtraction < Warehouse::Record
     dependent: :restrict_with_error,
     inverse_of: :financial_statement_extraction
 
+  has_many :financial_statement_line_items,
+    -> { order(:flow, :position) },
+    class_name: "Warehouse::FinancialStatementLineItem",
+    dependent: :restrict_with_error,
+    inverse_of: :financial_statement_extraction
+
   has_object :extractor
 
   validates :institution_canonical_id, :document_canonical_id, :asset_sha256,
     :fiscal_year_end, :extractor_version, :status, presence: true
   validates :asset_sha256, format: { with: /\A[0-9a-f]{64}\z/ },
-    uniqueness: { scope: [ :institution_release_id, :extractor_version ] }
+    uniqueness: { scope: [ :institution_release_id, :extractor_version, :fiscal_year_end ] }
   validates :status, inclusion: { in: STATUSES }
   validates :statement_basis, inclusion: { in: STATEMENT_BASES }
   validates :language, inclusion: { in: LANGUAGES }, allow_nil: true
   validate :review_fields_are_paired
+  validate :completed_extractions_have_check_results
+  validate :approved_extractions_have_review_provenance
   validate :source_asset_belongs_to_release
+  validate :document_year_matches_fiscal_year
 
   scope :approved, -> { where(status: "approved") }
+  scope :publishable_details, -> do
+    approved.where(extractor_version: Warehouse::FinancialStatementExtraction::DetailedPipeline::EXTRACTOR_VERSION)
+  end
   scope :for_institution, ->(canonical_id) { where(institution_canonical_id: canonical_id) }
 
+  def self.verification_checks(check_results)
+    Array(check_results).map do |check|
+      check = check.stringify_keys
+      { id: check["id"], status: check["status"], detail: check["detail"] }
+    end
+  end
+
   def approve!(reviewer:, notes: nil)
     raise ActiveRecord::RecordInvalid, self if reviewed_at.present?
     raise ArgumentError, "only extracted or needs-review results can be approved" unless status.in?(%w[extracted needs_review])
+    raise ArgumentError, "verification check results must be saved before approval" if check_results.blank?
 
     update!(status: "approved", reviewed_by: reviewer, reviewed_at: Time.current, review_notes: notes)
   end
@@ -47,6 +67,18 @@ def review_fields_are_paired
     errors.add(:reviewed_by, "must be present exactly when reviewed_at is present")
   end
 
+  def completed_extractions_have_check_results
+    return unless status.in?(%w[extracted needs_review approved rejected failed]) && check_results.blank?
+
+    errors.add(:check_results, "must be saved for a completed extraction")
+  end
+
+  def approved_extractions_have_review_provenance
+    return unless status == "approved" && (reviewed_at.blank? || reviewed_by.blank?)
+
+    errors.add(:reviewed_at, "and reviewer must be saved before approval")
+  end
+
   def source_asset_belongs_to_release
     return unless institution_release && document_canonical_id.present? && asset_sha256.present?
 
@@ -58,4 +90,14 @@ def source_asset_belongs_to_release
       errors.add(:asset_sha256, "must identify an archived asset on the release document")
     end
   end
+
+  def document_year_matches_fiscal_year
+    return if document_canonical_id.blank? || fiscal_year_end.blank?
+
+    match = document_canonical_id.match(%r{/documents/(?:financial-statements|annual-report)/(\d{4})/})
+    return unless match
+    return if match[1].to_i == fiscal_year_end.year
+
+    errors.add(:fiscal_year_end, "year must match the source document canonical ID")
+  end
 end