<commit_before>import argparse
import csv
import datetime
import sys

from colander import iso8601
from konfig import Config

from ichnaea.db import (
    Database,
    normalize_wifi_key,
)
from ichnaea.submit import process_measure


def load_file(session, source_file, batch_size=10000):
    utcnow = datetime.datetime.utcnow().replace(tzinfo=iso8601.UTC)
    utcmin = utcnow - datetime.timedelta(120)

    with open(source_file, 'r') as fd:
        reader = csv.reader(fd, delimiter='\t', quotechar=None)
        session_objects = []
        counter = 0

        for fields in reader:
            try:
                time = int(fields[0])
                if time == 0:  # pragma: no cover
                    # unknown time gets an old date
                    time = utcmin
                else:
                    # convert from unixtime to utc
                    time = datetime.datetime.utcfromtimestamp(time)

                key = normalize_wifi_key(str(fields[6]))
                if key == '000000000000':  # pragma: no cover
                    continue

                lat = fields[2]
                lon = fields[3]
                signal = int(fields[4])
                channel = int(fields[5])
                wifi = dict(
                    key=key,
                    channel=channel,
                    signal=signal,
                )
                data = dict(
                    lat=lat,
                    lon=lon,
                    time=time,
                    accuracy=0,
                    altitude=0,
                    altitude_accuracy=0,
                    radio='',
                    cell=(),
                    wifi=[wifi],
                )
            except (ValueError, IndexError):
                continue
            session_objects.extend(process_measure(data, utcnow, session))

            # flush every 1000 new records
            counter += 1
            if counter % batch_size == 0:
                session.add_all(session_objects)
                session.flush()
                print('Added %s records.' % counter)
                session_objects = []

    # add the rest
    session.add_all(session_objects)
    session.flush()
    return counter


def main(argv):
    parser = argparse.ArgumentParser(
        prog=argv[0], description='Location Importer')

    parser.add_argument('--dry-run', action='store_true')
    parser.add_argument('config', help="config file")
    parser.add_argument('source', help="source file")
    args = parser.parse_args(argv[1:])
    settings = Config(args.config).get_map('ichnaea')
    db = Database(
        settings['db_master'],
        socket=settings.get('db_master_socket'),
        create=False,
    )
    session = db.session()
    added = load_file(session, args.source)
    print('Added %s records.' % added)
    if args.dry_run:
        session.rollback()
    else:  # pragma: no cover
        session.commit()
    return added


def console_entry():  # pragma: no cover
    main(sys.argv)
<commit_msg>note a todo about the importer to avoid config file duplication<commit_after>import argparse
import csv
import datetime
import sys

from colander import iso8601
from konfig import Config

from ichnaea.db import (
    Database,
    normalize_wifi_key,
)
from ichnaea.submit import process_measure


def load_file(session, source_file, batch_size=10000):
    utcnow = datetime.datetime.utcnow().replace(tzinfo=iso8601.UTC)
    utcmin = utcnow - datetime.timedelta(120)

    with open(source_file, 'r') as fd:
        reader = csv.reader(fd, delimiter='\t', quotechar=None)
        session_objects = []
        counter = 0

        for fields in reader:
            try:
                time = int(fields[0])
                if time == 0:  # pragma: no cover
                    # unknown time gets an old date
                    time = utcmin
                else:
                    # convert from unixtime to utc
                    time = datetime.datetime.utcfromtimestamp(time)

                key = normalize_wifi_key(str(fields[6]))
                if key == '000000000000':  # pragma: no cover
                    continue

                lat = fields[2]
                lon = fields[3]
                signal = int(fields[4])
                channel = int(fields[5])
                wifi = dict(
                    key=key,
                    channel=channel,
                    signal=signal,
                )
                data = dict(
                    lat=lat,
                    lon=lon,
                    time=time,
                    accuracy=0,
                    altitude=0,
                    altitude_accuracy=0,
                    radio='',
                    cell=(),
                    wifi=[wifi],
                )
            except (ValueError, IndexError):
                continue
            session_objects.extend(process_measure(data, utcnow, session))

            # flush every 1000 new records
            counter += 1
            if counter % batch_size == 0:
                session.add_all(session_objects)
                session.flush()
                print('Added %s records.' % counter)
                session_objects = []

    # add the rest
    session.add_all(session_objects)
    session.flush()
    return counter


def main(argv):
    parser = argparse.ArgumentParser(
        prog=argv[0], description='Location Importer')

    parser.add_argument('--dry-run', action='store_true')
    # TODO rely on ICHNAEA_CFG / ichnaea.config, as the worker is relying
    # on it anyways
    parser.add_argument('config', help="config file")
    parser.add_argument('source', help="source file")
    args = parser.parse_args(argv[1:])
    settings = Config(args.config).get_map('ichnaea')
    db = Database(
        settings['db_master'],
        socket=settings.get('db_master_socket'),
        create=False,
    )
    session = db.session()
    added = load_file(session, args.source)
    print('Added %s records.' % added)
    if args.dry_run:
        session.rollback()
    else:  # pragma: no cover
        session.commit()
    return added


def console_entry():  # pragma: no cover
    main(sys.argv)

{'fraction_non_alphanumeric': 0.09486102133160956, 'fraction_numerical': 0.015190691661279896, 'mean_word_length': 1.614702154626109, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 2, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2928103', 'n_tokens_mistral': 1801, 'n_tokens_neox': 1638, 'n_words': 401}