<commit_before>from sqlalchemy import text
from sqlalchemy.exc import IntegrityError

from ichnaea.models import (
    Cell,
    CellMeasure,
    Measure,
    RADIO_TYPE,
    Wifi,
    WifiBlacklist,
    WifiMeasure,
)
from ichnaea.decimaljson import (
    decode_datetime,
    encode_datetime,
    loads,
)
from ichnaea.tasks import DatabaseTask
from ichnaea.worker import celery

from ichnaea.service.submit.utils import process_score

sql_null = None  # avoid pep8 warning


def create_cell_measure(measure_data, entry):
    return CellMeasure(
        measure_id=measure_data['id'],
        created=decode_datetime(measure_data.get('created', '')),
        lat=measure_data['lat'],
        lon=measure_data['lon'],
        time=decode_datetime(measure_data.get('time', '')),
        accuracy=measure_data.get('accuracy', 0),
        altitude=measure_data.get('altitude', 0),
        altitude_accuracy=measure_data.get('altitude_accuracy', 0),
        mcc=entry['mcc'],
        mnc=entry['mnc'],
        lac=entry.get('lac', 0),
        cid=entry.get('cid', 0),
        psc=entry.get('psc', 0),
        asu=entry.get('asu', 0),
        signal=entry.get('signal', 0),
        ta=entry.get('ta', 0),
    )


def update_cell_measure_count(measure, session, userid=None):
    if (measure.radio == -1 or measure.lac == 0 or measure.cid == 0):
        # only update data for complete records
        return

    # do we already know about these cells?
    query = session.query(Cell).filter(
        Cell.radio == measure.radio).filter(
        Cell.mcc == measure.mcc).filter(
        Cell.mnc == measure.mnc).filter(
        Cell.lac == measure.lac).filter(
        Cell.cid == measure.cid
    )
    cell = query.first()
    new_cell = 0
    if cell:
        if isinstance(cell.new_measures, (int, long)):
            cell.new_measures = Cell.new_measures + 1
        else:
            # already a sql expression
            cell.new_measures += 1
        if isinstance(cell.total_measures, (int, long)):
            if cell.total_measures < 5:
                # count cells as new until they show up in the search
                new_cell += 1
            cell.total_measures = Cell.total_measures + 1
        else:
            # already a sql expression
            cell.total_measures += 1
            if cell.total_measures.right.value < 5:
                # count cells as new until they show up in the search
                new_cell += 1
    else:
        cell = Cell(radio=measure.radio, mcc=measure.mcc, mnc=measure.mnc,
                    lac=measure.lac, cid=measure.cid,
                    new_measures=1, total_measures=1)
        new_cell += 1
        session.add(cell)
    if userid is not None and new_cell > 0:
        # update user score
        process_score(userid, new_cell, session, key='new_cell')


@celery.task(base=DatabaseTask, ignore_result=True)
def schedule_cell_cleanup(lower, upper, cell_lower, cell_upper):
    with schedule_cell_cleanup.db_session() as session:
        stmt = text("select id from measure where id > %s and id < %s and "
                    "cell is not null and id not in (select measure_id from "
                    "cell_measure where id > %s and id < %s)" % (
                        lower, upper, cell_lower, cell_upper))
        ids = session.execute(stmt).fetchall()
        ids = [int(i[0]) for i in ids]
        if ids:
            reprocess_cell_measure.delay(ids)
        return len(ids)
    return 0


@celery.task(base=DatabaseTask, ignore_result=True)
def reprocess_cell_measure(measure_ids, userid=None):
    measures = []
    try:
        with reprocess_cell_measure.db_session() as session:
            measures = session.query(Measure).filter(
                Measure.id.in_(measure_ids)).filter(
                Measure.cell != sql_null).all()
            for measure in measures:
                measure_data = dict(
                    id=measure.id, created=encode_datetime(measure.created),
                    lat=measure.lat, lon=measure.lon,
                    time=encode_datetime(measure.time),
                    accuracy=measure.accuracy, altitude=measure.altitude,
                    altitude_accuracy=measure.altitude_accuracy,
                    radio=measure.radio,
                )
                insert_cell_measure.delay(
                    measure_data, loads(measure.cell), userid=userid)
        return len(measures)
    except IntegrityError as exc:  # pragma: no cover
        # TODO log error
        return 0
    except Exception as exc:  # pragma: no cover
        raise reprocess_cell_measure.retry(exc=exc)


@celery.task(base=DatabaseTask, ignore_result=True)
def insert_cell_measure(measure_data, entries, userid=None):
    cell_measures = []
    try:
        with insert_cell_measure.db_session() as session:
            for entry in entries:
                cell = create_cell_measure(measure_data, entry)
                # use more specific cell type or
                # fall back to less precise measure
                if entry.get('radio'):
                    cell.radio = RADIO_TYPE.get(entry['radio'], -1)
                else:
                    cell.radio = measure_data['radio']
                update_cell_measure_count(cell, session, userid=userid)
                cell_measures.append(cell)
            session.add_all(cell_measures)
            session.commit()
        return len(cell_measures)
    except IntegrityError as exc:  # pragma: no cover
        # TODO log error
        return 0
    except Exception as exc:  # pragma: no cover
        raise insert_cell_measure.retry(exc=exc)<commit_msg>factor out `insert_cell_measure` logic into separate function<commit_after>from sqlalchemy import text
from sqlalchemy.exc import IntegrityError

from ichnaea.models import (
    Cell,
    CellMeasure,
    Measure,
    RADIO_TYPE,
    Wifi,
    WifiBlacklist,
    WifiMeasure,
)
from ichnaea.decimaljson import (
    decode_datetime,
    encode_datetime,
    loads,
)
from ichnaea.tasks import DatabaseTask
from ichnaea.worker import celery

from ichnaea.service.submit.utils import process_score

sql_null = None  # avoid pep8 warning


def create_cell_measure(measure_data, entry):
    return CellMeasure(
        measure_id=measure_data['id'],
        created=decode_datetime(measure_data.get('created', '')),
        lat=measure_data['lat'],
        lon=measure_data['lon'],
        time=decode_datetime(measure_data.get('time', '')),
        accuracy=measure_data.get('accuracy', 0),
        altitude=measure_data.get('altitude', 0),
        altitude_accuracy=measure_data.get('altitude_accuracy', 0),
        mcc=entry['mcc'],
        mnc=entry['mnc'],
        lac=entry.get('lac', 0),
        cid=entry.get('cid', 0),
        psc=entry.get('psc', 0),
        asu=entry.get('asu', 0),
        signal=entry.get('signal', 0),
        ta=entry.get('ta', 0),
    )


def update_cell_measure_count(measure, session, userid=None):
    if (measure.radio == -1 or measure.lac == 0 or measure.cid == 0):
        # only update data for complete records
        return

    # do we already know about these cells?
    query = session.query(Cell).filter(
        Cell.radio == measure.radio).filter(
        Cell.mcc == measure.mcc).filter(
        Cell.mnc == measure.mnc).filter(
        Cell.lac == measure.lac).filter(
        Cell.cid == measure.cid
    )
    cell = query.first()
    new_cell = 0
    if cell:
        if isinstance(cell.new_measures, (int, long)):
            cell.new_measures = Cell.new_measures + 1
        else:
            # already a sql expression
            cell.new_measures += 1
        if isinstance(cell.total_measures, (int, long)):
            if cell.total_measures < 5:
                # count cells as new until they show up in the search
                new_cell += 1
            cell.total_measures = Cell.total_measures + 1
        else:
            # already a sql expression
            cell.total_measures += 1
            if cell.total_measures.right.value < 5:
                # count cells as new until they show up in the search
                new_cell += 1
    else:
        cell = Cell(radio=measure.radio, mcc=measure.mcc, mnc=measure.mnc,
                    lac=measure.lac, cid=measure.cid,
                    new_measures=1, total_measures=1)
        new_cell += 1
        session.add(cell)
    if userid is not None and new_cell > 0:
        # update user score
        process_score(userid, new_cell, session, key='new_cell')


@celery.task(base=DatabaseTask, ignore_result=True)
def schedule_cell_cleanup(lower, upper, cell_lower, cell_upper):
    with schedule_cell_cleanup.db_session() as session:
        stmt = text("select id from measure where id > %s and id < %s and "
                    "cell is not null and id not in (select measure_id from "
                    "cell_measure where id > %s and id < %s)" % (
                        lower, upper, cell_lower, cell_upper))
        ids = session.execute(stmt).fetchall()
        ids = [int(i[0]) for i in ids]
        if ids:
            reprocess_cell_measure.delay(ids)
        return len(ids)
    return 0


@celery.task(base=DatabaseTask, ignore_result=True)
def reprocess_cell_measure(measure_ids, userid=None):
    measures = []
    try:
        with reprocess_cell_measure.db_session() as session:
            measures = session.query(Measure).filter(
                Measure.id.in_(measure_ids)).filter(
                Measure.cell != sql_null).all()
            for measure in measures:
                measure_data = dict(
                    id=measure.id, created=encode_datetime(measure.created),
                    lat=measure.lat, lon=measure.lon,
                    time=encode_datetime(measure.time),
                    accuracy=measure.accuracy, altitude=measure.altitude,
                    altitude_accuracy=measure.altitude_accuracy,
                    radio=measure.radio,
                )
                insert_cell_measure.delay(
                    measure_data, loads(measure.cell), userid=userid)
        return len(measures)
    except IntegrityError as exc:  # pragma: no cover
        # TODO log error
        return 0
    except Exception as exc:  # pragma: no cover
        raise reprocess_cell_measure.retry(exc=exc)


def process_cell_measure(session, measure_data, entries, userid=None):
    cell_measures = []
    for entry in entries:
        cell = create_cell_measure(measure_data, entry)
        # use more specific cell type or
        # fall back to less precise measure
        if entry.get('radio'):
            cell.radio = RADIO_TYPE.get(entry['radio'], -1)
        else:
            cell.radio = measure_data['radio']
        update_cell_measure_count(cell, session, userid=userid)
        cell_measures.append(cell)
    session.add_all(cell_measures)
    return cell_measures


@celery.task(base=DatabaseTask, ignore_result=True)
def insert_cell_measure(measure_data, entries, userid=None):
    try:
        cell_measures = []
        with insert_cell_measure.db_session() as session:
            cell_measures = process_cell_measure(
                session, measure_data, entries, userid=userid)
            session.commit()
        return len(cell_measures)
    except IntegrityError as exc:  # pragma: no cover
        # TODO log error
        return 0
    except Exception as exc:  # pragma: no cover
        raise insert_cell_measure.retry(exc=exc)
{'fraction_non_alphanumeric': 0.09769512833944474, 'fraction_numerical': 0.0054129561725161516, 'mean_word_length': 2.2358757062146895, 'pattern_counts': {'":': 0, '<': 11, '<?xml version=': 0, '>': 9, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, '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': '7126344', 'n_tokens_mistral': 3434, 'n_tokens_neox': 3270, 'n_words': 797}