def update_network_entities(engine, file_name):
    log.info("Updating network entities reference sheet: %s", file_name)
    network_entities = set()
    table = sl.get_table(engine, 'network_entity')
    if os.path.exists(file_name):
        fh = open(file_name, 'rb')
        reader = csv.DictReader(fh)
        for d in reader:
            e = [(k, v.decode('utf-8')) for (k, v) in d.items()]
            e = dict(e)
            network_entities.add((e['representativeEtlId'], e['etlFingerPrint']))
            sl.upsert(engine, table, e, ['representativeEtlId', 'etlFingerPrint'])
        fh.close()
        reps = set([ne[0] for ne in network_entities])
        rep_table = sl.get_table(engine, 'representative')
        for rep in reps:
            sl.update(engine, rep_table, {'etlId': rep}, {'network_extracted': True})

    for row in sl.all(engine, table):
        network_entities.add((row['representativeEtlId'], row['etlFingerPrint']))

    fh = open(file_name, 'wb')
    writer = None
    table = sl.get_table(engine, 'network_entity')
    for ic, fp in network_entities:
        row = {
            'representativeEtlId': ic,
            'etlFingerPrint': fp
        }
        if writer is None:
            writer = csv.DictWriter(fh, row.keys())
            writer.writerow(dict(zip(row.keys(), row.keys())))
        r = [(k, unicode(v).encode('utf-8')) for k, v in row.items()]
        writer.writerow(dict(r))
    fh.close()

def load_debatten(engine, sitzung):
    WebTV_Speech = sl.get_table(engine, 'webtv_speech')
    zitate = list(sl.find(engine, WebTV_Speech, wp=str(sitzung.wahlperiode),
        session=str(sitzung.nummer)))
    debatten = dict([(z['item_id'], z) for z in zitate])
    speeches = list(sl.find(engine, sl.get_table(engine, 'speech'),
        wahlperiode=int(sitzung.wahlperiode), sitzung=int(sitzung.nummer)))
    for i, data in debatten.items():
        log.info("Loading  -> Debatte: %s..." % data.get('item_label'))
        debatte = Debatte.query.filter_by(
                sitzung=sitzung,
                nummer=data.get('item_id')
                ).first()
        if debatte is None:
            debatte = Debatte()
        debatte.sitzung = sitzung
        debatte.nummer = data.get('item_id')
        debatte.tops = data.get('item_key')
        debatte.titel = data.get('item_label')
        debatte.text = data.get('item_description')

        db.session.add(debatte)
        db.session.flush()

        dzitate = filter(lambda z: z['item_id'] == data['item_id'], zitate)
        load_zitate(engine, debatte, dzitate, speeches)
        db.session.commit()

def make_fingerprint(engine, person):
    try:
        long_name = make_long_name(person)
        try:
            long_name = resolve_person(long_name)
            log.info(" -> %s" % long_name.strip())
        except:
            log.error("Resolve did not work")
            pass

        Person = sl.get_table(engine, 'person')
        sl.upsert(engine, Person, {
            'fingerprint': long_name,
            'slug': url_slug(long_name),
            'mdb_id': person['mdb_id']
            }, unique=['mdb_id'])
        Rolle = sl.get_table(engine, 'rolle')
        sl.upsert(engine, Rolle, {
            'mdb_id': person['mdb_id'],
            'fingerprint': long_name
            }, unique=['mdb_id'])
        person['fingerprint'] = long_name
    except BadReference:
        log.error("Bad Reference %s", person)
        pass

def load_finances(financialData, childBase, engine):
    etlId = '%s//%s' % (financialData['startDate'].isoformat(),
                        financialData['endDate'].isoformat())

    financial_sources = \
        [(s, 'other') for s in financialData.pop("otherCustomized")] + \
        [(s, 'public') for s in financialData.pop("publicCustomized")]
    for financial_source, type_ in financial_sources:
        financial_source['type'] = type_
        financial_source['financialDataEtlId'] = etlId
        financial_source.update(childBase)
        sl.upsert(engine, sl.get_table(engine, 'financialDataCustomSource'),
                  financial_source, ['representativeEtlId',
                      'financialDataEtlId', 'type', 'name'])

    for turnover in financialData.pop("turnoverBreakdown"):
        turnover['financialDataEtlId'] = etlId
        turnover['etlFingerPrint'] = turnover['name'].strip()
        turnover.update(childBase)
        sl.upsert(engine, sl.get_table(engine, 'financialDataTurnover'),
                  turnover, ['representativeEtlId', 'financialDataEtlId',
                             'etlFingerPrint'])

    financialData['etlId'] = etlId
    financialData.update(childBase)
    sl.upsert(engine, sl.get_table(engine, 'financialData'),
              financialData, ['representativeEtlId', 'etlId'])

def load_ablauf(engine, indexer, data):
    ablauf = Ablauf.query.filter_by(source_url=data.get('source_url')).first()
    if ablauf is None:
        ablauf = Ablauf()

    ablauf.key = data.get('key')
    ablauf.source_url = data.get('source_url')
    ablauf.wahlperiode = data.get('wahlperiode')
    ablauf.typ = data.get('typ')
    ablauf.klasse = data.get('class')
    ablauf.titel = data.get('titel')
    if not len(ablauf.titel):
        log.error("No titel!")
        return

    ablauf.initiative = data.get('initiative')
    ablauf.stand = data.get('stand')
    ablauf.signatur = data.get('signatur')
    ablauf.gesta_id = data.get('gesta_id')
    ablauf.eu_dok_nr = data.get('eu_dok_nr')
    ablauf.eur_lex_url = data.get('eur_lex_url')
    ablauf.eur_lex_pdf = data.get('eur_lex_pdf')
    ablauf.consilium_url = data.get('consilium_url')
    ablauf.abstrakt = data.get('abstrakt')
    ablauf.zustimmungsbeduerftig = data.get('zustimmungsbeduerftig')
    ablauf.sachgebiet = data.get('sachgebiet')
    ablauf.abgeschlossen = True if str(data.get('abgeschlossen')) \
            == 'True' else False
    db.session.add(ablauf)
    db.session.flush()

    worte = []
    _Schlagwort = sl.get_table(engine, 'schlagwort')
    for sw in sl.find(engine, _Schlagwort, source_url=ablauf.source_url):
        wort = Schlagwort()
        wort.name = sw['wort']
        db.session.add(wort)
        worte.append(wort)
    ablauf.schlagworte = worte

    _Referenz = sl.get_table(engine, 'referenz')
    for ddata in sl.find(engine, _Referenz, source_url=ablauf.source_url):
        dokument = load_dokument(engine, indexer, ddata)
        referenz = Referenz.query.filter_by(
                dokument=dokument,
                seiten=ddata.get('seiten'),
                ).filter(Referenz.ablaeufe.any(id=ablauf.id)).first()
        if referenz is None:
            referenz = Referenz()
            referenz.ablaeufe.append(ablauf)
            referenz.dokument = dokument
        referenz.seiten = ddata.get('seiten')
        referenz.text = ddata.get('text')

    _Position = sl.get_table(engine, 'position')
    for position in sl.find(engine, _Position, source_url=ablauf.source_url):
        load_position(engine, indexer, ablauf, position)

    db.session.commit()
    indexer.add(ablauf)

def supplier_names_from_table(engine, resource_id, table_id):
    global tables_count, rows_count
    tables_count = tables_count + 1
    print "# checking table %d" % tables_count
    table_suffix = "%s_table%s" % (resource_id, table_id)

    table = sl.get_table(engine, "spending_%s" % table_suffix)
    supplier_table = sl.get_table(engine, "suppliers")

    for row in sl.all(engine, table):
        rows_count = rows_count + 1
        if not row.has_key("SupplierName"):
            # One of the junk tables that contain no real data, usually excel noise
            continue
        supplier = row["SupplierName"]

        if supplier is None or supplier == "":
            continue

        if suppliers_visited.has_key(supplier):
            continue

        suppliers_visited[supplier] = True

        if sl.find_one(engine, supplier_table, original=supplier) is not None:
            continue

        yield supplier

def condense(engine, resource_id, table_id, force):
    table_suffix = '%s_table%s' % (resource_id, table_id)

    if not engine.has_table('raw_%s' % table_suffix):
        return

    condensed_table = sl.get_table(engine, 'condensed')

    # Skip over tables we have already extracted
    if not force and sl.find_one(engine, condensed_table, resource_id=resource_id, table_id=table_id) is not None:
        return

    connection = engine.connect()
    trans = connection.begin()

    start = time.time()

    try:
        raw_table = sl.get_table(connection, 'raw_%s' % table_suffix)
        sl.drop_table(connection, 'spending_%s' % table_suffix)
        spending_table = sl.get_table(connection, 'spending_%s' % table_suffix)
        columns_table = sl.get_table(connection, 'column_sets')

        normalise_map = normalised_columns_map(raw_table)
        normalised_headers = ','.join(sorted(normalise_map.values()))
        mapping_row = sl.find_one(connection, columns_table, normalised=normalised_headers)

        if mapping_row is None or not mapping_row.get('valid'):
            # This table is unmapped, cannot be condensed
            return

        column_mapping = json.loads(mapping_row['column_map'])

        # Build the final mapping from input column to output column
        mapping = {}
        for k,n in normalise_map.iteritems():
            if n in column_mapping and column_mapping[n] is not None and len(column_mapping[n]) > 0:
                mapping[k] = column_mapping[n]
        
        for row in sl.all(connection, raw_table):
            spending_row = {}
            for key, value in row.items():
                if key not in mapping:
                    continue
                if not value or not len(value.strip()):
                    continue
                if mapping[key] in spending_row:
                    continue
                spending_row[mapping[key]] = value.strip()
            #print spending_row
            sl.add_row(connection, spending_table, spending_row)
        sl.upsert(connection, condensed_table, {'resource_id': resource_id,
                                                'table_id': table_id,
                                                'condense_time': time.time() - start,
                                                }, ['resource_id', 'table_id'])
        trans.commit()
    finally:
        connection.close()

def speechmatcher(wp, session):
    engine = etl_engine()
    speech_table = sl.get_table(engine, 'speech')
    speeches = sl.find(engine, speech_table, order_by='sequence', 
        wahlperiode=wp, sitzung=session, matched=True)
    webtv_table = sl.get_table(engine, 'webtv')
    agenda = sl.find(engine, webtv_table, wp=wp, session=session)
    agenda = list(agenda)
    return render_template('backend/speechmatcher.html',
            speeches=speeches, agenda=agenda, wp=wp, session=session)

def create_entities(engine):
    log.info("De-normalizing global entities collection...")
    table = sl.get_table(engine, 'entity')
    for tbl in ['representative', 'person', 'financialDataTurnover',
        'organisation', 'network_entity']:
        for row in sl.all(engine, sl.get_table(engine, tbl)):
            entity = {'etlFingerPrint': row.get('etlFingerPrint')}
            entity['legalStatus'] = row.get('legalStatus', '')
            entity['countryCode'] = row.get('contactCountryCode', '')
            entity['etlTable'] = tbl
            sl.upsert(engine, table, entity, ['etlFingerPrint', 'etlTable'])

def add_to_gremium(node, url, role, engine):
    key = node.get('id')
    table = sl.get_table(engine, 'gremium')
    g = sl.find_one(engine, table, key=key)
    if g is None:
        g = {'key': key, 'type': 'sonstiges'}
        g['name'] = node.findtext('gremiumName')
        g['url'] = node.findtext('gremiumURL')
        sl.upsert(engine, table, g, unique=['key'])
    table = sl.get_table(engine, 'gremium_mitglieder')
    sl.upsert(engine, table, {
        'gremium_key': g['key'],
        'person_source_url': url,
        'role': role
        }, unique=['person_source_url', 'gremium_key', 'role'])

def scrape_transcript(engine, url, force=False):
    wp, session = url_metadata(url)
    table = sl.get_table(engine, 'speech')
    sample = sl.find_one(engine, table, source_url=url, matched=True)
    response, sio = fetch_stream(url)
    sample = check_tags(sample or {}, response, force)
    base_data = {'source_url': url, 
                 'sitzung': session,
                 'wahlperiode': wp,
                 'matched': False,
                 'loaded': False,
                 'source_etag': sample['source_etag']}
    log.info("Loading transcript: %s/%s, from %s" , wp, session, url)
    seq = 0
    parser = SpeechParser(sio)
    for contrib in parser:
        if not len(contrib['text'].strip()):
            continue
        contrib.update(base_data)
        contrib['sequence'] = seq
        sl.upsert(engine, table, contrib, 
                  unique=['source_url', 'sequence'])
        seq += 1
    if not parser.missing_recon:
        sl.upsert(engine, table, {
                    'matched': True,
                    'source_url': url,
            }, unique=['source_url'])
    else:
        raise InvalidReference()
    return base_data

def load_rollen(engine, person, data):
    _RolleSource = sl.get_table(engine, "rolle")
    for rdata in sl.find(engine, _RolleSource, fingerprint=data["fingerprint"]):
        rolle = Rolle.query.filter_by(
            person=person,
            funktion=rdata.get("funktion"),
            ressort=rdata.get("ressort"),
            fraktion=rdata.get("fraktion"),
            land=rdata.get("land"),
        ).first()
        if rolle is None:
            rolle = Rolle()

        rolle.person = person
        rolle.mdb_id = rdata.get("mdb_id")
        rolle.status = rdata.get("status")
        rolle.funktion = rdata.get("funktion")
        rolle.fraktion = rdata.get("fraktion")
        rolle.gewaehlt = rdata.get("gewaehlt")
        rolle.ressort = rdata.get("ressort")
        rolle.land = rdata.get("land")
        rolle.austritt = to_date(rdata.get("austritt"))

        if rdata.get("mdb_id"):
            rolle.wahlkreis = load_wahlkreis(engine, rolle, data)
        db.session.add(rolle)

def load(engine, grano):
    for rep in sl.find(engine, sl.get_table(engine, 'representative')):
        del rep['id']
        rep_ent = canonical_actor(grano, engine, rep['originalName'])
        if 'id' in rep_ent:
            rep_ent = grano.getEntity(rep_ent['id'], deep=True)
        #if not SETTINGS.FULL and rep_ent['etlId'] == rep['etlId']:
        #    continue
        rep_ent.update(rep)
        rep_ent['actsAsRepresentative'] = True
        rep_ent['staffMembers'] = int(float(rep['members']))
        rep_ent['incoming'] = rep_ent.get('incoming', [])
        rep_ent['outgoing'] = rep_ent.get('outgoing', [])
        rep_ent['contactCountry'] = rep_ent['contactCountryNorm']
        rep_ent = load_clients(grano, engine, rep_ent)
        rep_ent = load_organisations(grano, engine, rep_ent)
        rep_ent = load_networking(grano, engine, rep_ent)
        rep_ent = load_persons(grano, engine, rep_ent)
        rep_ent = load_interests(grano, engine, rep_ent)
        rep_ent = load_action_fields(grano, engine, rep_ent)
        rep_ent = get_financial_data(engine, rep_ent)
        # TODO: other financial sources
        #from pprint import pprint
        #pprint(rep_ent)
        grano.updateEntity(rep_ent)

def load_abstimmung(engine, source_url):
    table = sl.get_table(engine, 'abstimmung')
    stimmen = list(sl.find(engine, table, source_url=source_url,
        matched=True))
    if not len(stimmen):
        log.error("No reconciled votes, signals deeper trouble?")
        return
    thema = stimmen[0].get('subject')
    abst = Abstimmung.query.filter_by(thema=thema).first()
    if abst is None:
        abst = Abstimmung()
        abst.thema = thema
        abst.datum = to_date(stimmen[0].get('date'))
    db.session.add(abst)
    db.session.flush()
    for stimme_ in stimmen:
        person = Person.query.filter_by(
            fingerprint=stimme_.get('fingerprint')).first()
        if person is None:
            continue
        stimme = Stimme.query.filter_by(
            abstimmung=abst).filter_by(
            person=person).first()
        if stimme is not None:
            continue
        stimme = Stimme()
        stimme.entscheidung = stimme_['vote']
        stimme.person = person
        stimme.abstimmung = abst
        db.session.add(stimme)
    db.session.commit()

def load_rollen(engine, person, data):
    _RolleSource = sl.get_table(engine, 'rolle')
    mdb_rolle = None
    for rdata in sl.find(engine, _RolleSource, fingerprint=data['fingerprint']):
        rolle = Rolle.query.filter_by(
                person=person,
                funktion=rdata.get('funktion'),
                ressort=rdata.get('ressort'),
                fraktion=rdata.get('fraktion'),
                land=rdata.get('land')).first()
        if rolle is None:
            rolle = Rolle()

        rolle.person = person
        rolle.mdb_id = rdata.get('mdb_id')
        rolle.status = rdata.get('status')
        rolle.funktion = rdata.get('funktion')
        rolle.fraktion = rdata.get('fraktion')
        rolle.gewaehlt = rdata.get('gewaehlt')
        rolle.ressort = rdata.get('ressort')
        rolle.land = rdata.get('land')
        rolle.austritt = date(rdata.get('austritt'))

        if rdata.get('mdb_id'):
            rolle.wahlkreis = load_wahlkreis(engine, rolle, data)
            mdb_rolle = rolle
        db.session.add(rolle)
    return mdb_rolle

def generate_person_long_names(engine):
    log.info("Generating person fingerprints and slugs...")
    from offenesparlament.transform.namematch import match_speaker
    nkp = nk_persons()
    Person = sl.get_table(engine, 'person')
    for person in sl.find(engine, Person):
        long_name = make_long_name(person)
        try:
            long_name = match_speaker(long_name)
        except NKNoMatch:
            pass
        log.info(" -> %s" % long_name.strip())
        slug = url_slug(long_name)
        sl.upsert(engine, Person, {
                         'fingerprint': long_name,
                         'slug': slug,
                         'id': person['id']},
                         unique=['id'])
        tries = 0
        while True:
            try:
                nkp.ensure_value(long_name, data=person)
            except ValueError, E:
                log.warn('Exception: %s' % str(E))
                tries = tries + 1
                if tries > 5:
                    raise
            else:
                break

def load_news(engine):
    log.info("Loading news into production DB...")
    _NewsSource = sl.get_table(engine, 'news')
    for data in sl.all(engine, _NewsSource):
        news = NewsItem.query.filter_by(
                source_url=data.get('source_url')).first()
        if news is None:
            news = NewsItem()
        news.key = data.get('key')
        news.source_url = data.get('source_url')
        news.title = data.get('title')
        news.text = data.get('text')
        news.date = date(data.get('date'))
        news.image_url = data.get('image_url')
        news.image_copyright = data.get('image_copyright')
        news.image_changed_at = date(data.get('image_changed_at'))

        if data.get('gremium'):
            gremium = Gremium.query.filter_by(
                    key=data.get('gremium')).first()
            if gremium is None:
                log.error("Gremium %s not found!" % data.get('gremium'))
            else:
                news.gremium = gremium
        db.session.add(news)
    db.session.commit()

def combine(force=False, filter=None):
    stats = OpenSpendingStats()
    engine = db_connect()
    source_table = sl.get_table(engine, 'source')
    for row in sl.find(engine, source_table, **(filter or {})):
        combine_resource(engine, source_table, row, force, stats)
    log.info('Combine summary: \n%s' % stats.report())

def combine_resource_core(engine, row, stats):
    '''Given a resource (source row) it opens the table with its contents
    and puts its rows into the combined table, using the column mappings.

    Returns whether it succeeds or not.
    '''
    error = None
    for sheet_id in range(0, row['sheets']):
        table = sl.get_table(engine, 'raw_%s_sheet%s' % (
            row['resource_id'], sheet_id))
        if not engine.has_table(table.name):
            error = 'Sheet table does not exist'
            log.warn('Sheet table does not exist: %s', table)
            continue
        columns = [c.name for c in table.columns]
        mapping = column_mapping(engine, row, columns)
        if mapping is None:
            error = 'Column mappings not complete'
            log.warn('Column mappings not complete: %s', columns)
            continue
        log.debug('Column mapping: %r', mapping)
        if not combine_sheet(engine, row, sheet_id, table, mapping):
            error = 'Could not combine sheet'
    if error:
        stats.add_source(error, row)
    else:
        stats.add_source('Combined ok', row)
    return (not error)

def combine_sheet(engine, resource, sheet_id, table, mapping):
    begin = time.time()
    base = {
            'resource_id': resource['resource_id'],
            'resource_hash': resource['extract_hash'],
            'sheet_id': sheet_id,
        }
    spending_table = sl.get_table(engine, 'spending')
    connection = engine.connect()
    trans = connection.begin()
    try:
        rows = 0
        sl.delete(connection, spending_table,
                resource_id=resource['resource_id'],
                sheet_id=sheet_id)
        for row in sl.all(connection, table):
            data = dict(base)
            for col, value in row.items():
                if col == 'id':
                    data['row_id'] = value
                    continue
                mapped = mapping.get(col)
                if mapped is not None:
                    data[mapped] = value
            sl.add_row(connection, spending_table, data)
            rows += 1
        trans.commit()
        log.info("Loaded %s rows in %s ms", rows,
                int((time.time()-begin)*1000))
        return rows > 0
    finally:
        connection.close()