<filename>af/prio/prio.py
#!/usr/bin/env python3
# -*- coding: utf-8 -*-

"""OONI URL prioritization service

Configured with /etc/ooni/prio.conf

Runs as a system daemon

Inputs:
    Database tables:
        `citizenlab` db table
        country codes are in the `cc` column, lowercase, with the exception of ZZ

Outputs:
    Files in /var/lib/analysis
    Node exporter / prometheus metrics
    Dedicated unlogged database tables and charts
        tables:
            currently_blocked

Special country code values:
    ZZ: unknown
    XX: all
"""

from collections import namedtuple
from configparser import ConfigParser
from typing import List
import logging
import random
import time

from systemd.journal import JournalHandler  # debdeps: python3-systemd

from bottle import route
import bottle

from psycopg2.extras import RealDictCursor
import psycopg2

conf = None
test_items = {}
last_update_time = 0

log = logging.getLogger("prio")
log.addHandler(JournalHandler(SYSLOG_IDENTIFIER="prio"))


def connect_db(c):
    conn = psycopg2.connect(
        dbname=c.dbname,
        user=c.dbuser,
        host=c.dbhost,
        port=c.dbport,
        password=<PASSWORD>,
    )
    return conn


# @metrics.timer("update_url_prioritization")
def update_url_prioritization():
    """
    """
    log.info("Started update_url_prioritization")
    conn = connect_db(conf)
    cur = conn.cursor(cursor_factory=RealDictCursor)

    log.info("Regenerating URL prioritization file")
    sql = """SELECT priority, domain, url, cc, category_code FROM citizenlab"""
    cur.execute(sql)
    entries = list(cur.fetchall())
    conn.rollback()
    conn.close()

    # Create dict: cc -> category_code -> [entry, ... ]
    entries_by_country = {}
    for e in entries:
        country = e["cc"].upper()
        if country not in entries_by_country:
            entries_by_country[country] = {}
        ccode = e["category_code"]
        entries_by_country[country].setdefault(ccode, []).append(e)

    # merge ZZ into each country
    zz = entries_by_country.pop("ZZ")
    for ccode, country_dict in entries_by_country.items():
        for category_code, test_items in zz.items():
            country_dict.setdefault(category_code, []).extend(test_items)

    log.info("Update done")
    return entries_by_country


def algo_chao(s: List, k: int) -> List:
    """Chao weighted random sampling
    """
    n = len(s)
    assert len(s) >= k
    wsum = 0
    r = s[:k]
    assert len(r) == k
    for i in range(0, n):
        wsum = wsum + s[i]["priority"]
        if i < k:
            continue
        p = s[i]["priority"] / wsum  # probability for this item
        j = random.random()
        if j <= p:
            pos = random.randint(0, k - 1)
            r[pos] = s[i]

    return r


def generate_test_list(country_code: str, category_codes: str, limit: int):
    global test_items, last_update_time

    if last_update_time < time.time() - 100:  # conf.refresh_interval:
        last_update_time = time.time()
        try:
            test_items = update_url_prioritization()
        except Exception as e:
            log.error(e, exc_info=1)

    candidates_d = test_items[country_code]  # category_code -> [test_item, ... ]

    if category_codes:
        category_codes = [c.strip().upper() for c in category_codes.split(",")]
    else:
        category_codes = candidates_d.keys()

    candidates = []
    for ccode in category_codes:
        s = candidates_d.get(ccode, [])
        candidates.extend(s)

    log.info("%d candidates", len(candidates))

    if limit == -1:
        limit = 100
    limit = min(limit, len(candidates))
    selected = algo_chao(candidates, limit)

    out = []
    for entry in selected:
        out.append(
            {
                "category_code": entry["category_code"],
                "url": entry["url"],
                "country_code": "XX" if entry["cc"] == "ZZ" else entry["cc"],
            }
        )
    return out


@route("/api/v1/test-list/urls")
def list_urls():
    """
    https://orchestrate.ooni.io/api/v1/test-list/urls?country_code=IT
    """
    try:
        country_code = bottle.request.query.country_code.upper() or "ZZ"
        category_codes = bottle.request.query.category_code
        limit = int(bottle.request.query.limit or -1)
        test_items = generate_test_list(country_code, category_codes, limit)
        out = {
            "metadata": {
                "count": len(test_items),
                "current_page": -1,
                "limit": -1,
                "next_url": "",
                "pages": 1,
            },
            "results": test_items,
        }
        return out
    except Exception as e:
        log.error(e, exc_info=1)
        return {}


def main():
    global conf
    conffile = "/etc/ooni/prio.conf"
    cp = ConfigParser()
    with open(conffile) as f:
        cp.read_file(f)
    d = cp.defaults()  # parsed values from DEFAULT section
    conf = namedtuple("Conf", d.keys())(*d.values())
    bottle.run(host="localhost", port=conf.apiport)


if __name__ == "__main__":
    main()

<filename>af/prio/prio.py
#!/usr/bin/env python3
# -*- coding: utf-8 -*-

"""OONI URL prioritization service

Configured with /etc/ooni/prio.conf

Runs as a system daemon

Inputs:
    Database tables:
        `citizenlab` db table
        country codes are in the `cc` column, lowercase, with the exception of ZZ

Outputs:
    Files in /var/lib/analysis
    Node exporter / prometheus metrics
    Dedicated unlogged database tables and charts
        tables:
            currently_blocked

Special country code values:
    ZZ: unknown
    XX: all
"""

from collections import namedtuple
from configparser import ConfigParser
from typing import List
import logging
import random
import time

from systemd.journal import JournalHandler  # debdeps: python3-systemd

from bottle import route
import bottle

from psycopg2.extras import RealDictCursor
import psycopg2

conf = None
test_items = {}
last_update_time = 0

log = logging.getLogger("prio")
log.addHandler(JournalHandler(SYSLOG_IDENTIFIER="prio"))


def connect_db(c):
    conn = psycopg2.connect(
        dbname=c.dbname,
        user=c.dbuser,
        host=c.dbhost,
        port=c.dbport,
        password=<PASSWORD>,
    )
    return conn


# @metrics.timer("update_url_prioritization")
def update_url_prioritization():
    """
    """
    log.info("Started update_url_prioritization")
    conn = connect_db(conf)
    cur = conn.cursor(cursor_factory=RealDictCursor)

    log.info("Regenerating URL prioritization file")
    sql = """SELECT priority, domain, url, cc, category_code FROM citizenlab"""
    cur.execute(sql)
    entries = list(cur.fetchall())
    conn.rollback()
    conn.close()

    # Create dict: cc -> category_code -> [entry, ... ]
    entries_by_country = {}
    for e in entries:
        country = e["cc"].upper()
        if country not in entries_by_country:
            entries_by_country[country] = {}
        ccode = e["category_code"]
        entries_by_country[country].setdefault(ccode, []).append(e)

    # merge ZZ into each country
    zz = entries_by_country.pop("ZZ")
    for ccode, country_dict in entries_by_country.items():
        for category_code, test_items in zz.items():
            country_dict.setdefault(category_code, []).extend(test_items)

    log.info("Update done")
    return entries_by_country


def algo_chao(s: List, k: int) -> List:
    """Chao weighted random sampling
    """
    n = len(s)
    assert len(s) >= k
    wsum = 0
    r = s[:k]
    assert len(r) == k
    for i in range(0, n):
        wsum = wsum + s[i]["priority"]
        if i < k:
            continue
        p = s[i]["priority"] / wsum  # probability for this item
        j = random.random()
        if j <= p:
            pos = random.randint(0, k - 1)
            r[pos] = s[i]

    return r


def generate_test_list(country_code: str, category_codes: str, limit: int):
    global test_items, last_update_time

    if last_update_time < time.time() - 100:  # conf.refresh_interval:
        last_update_time = time.time()
        try:
            test_items = update_url_prioritization()
        except Exception as e:
            log.error(e, exc_info=1)

    candidates_d = test_items[country_code]  # category_code -> [test_item, ... ]

    if category_codes:
        category_codes = [c.strip().upper() for c in category_codes.split(",")]
    else:
        category_codes = candidates_d.keys()

    candidates = []
    for ccode in category_codes:
        s = candidates_d.get(ccode, [])
        candidates.extend(s)

    log.info("%d candidates", len(candidates))

    if limit == -1:
        limit = 100
    limit = min(limit, len(candidates))
    selected = algo_chao(candidates, limit)

    out = []
    for entry in selected:
        out.append(
            {
                "category_code": entry["category_code"],
                "url": entry["url"],
                "country_code": "XX" if entry["cc"] == "ZZ" else entry["cc"],
            }
        )
    return out


@route("/api/v1/test-list/urls")
def list_urls():
    """
    https://orchestrate.ooni.io/api/v1/test-list/urls?country_code=IT
    """
    try:
        country_code = bottle.request.query.country_code.upper() or "ZZ"
        category_codes = bottle.request.query.category_code
        limit = int(bottle.request.query.limit or -1)
        test_items = generate_test_list(country_code, category_codes, limit)
        out = {
            "metadata": {
                "count": len(test_items),
                "current_page": -1,
                "limit": -1,
                "next_url": "",
                "pages": 1,
            },
            "results": test_items,
        }
        return out
    except Exception as e:
        log.error(e, exc_info=1)
        return {}


def main():
    global conf
    conffile = "/etc/ooni/prio.conf"
    cp = ConfigParser()
    with open(conffile) as f:
        cp.read_file(f)
    d = cp.defaults()  # parsed values from DEFAULT section
    conf = namedtuple("Conf", d.keys())(*d.values())
    bottle.run(host="localhost", port=conf.apiport)


if __name__ == "__main__":
    main()

#!/usr/bin/env python3 # -*- coding: utf-8 -*- OONI URL prioritization service  Configured with /etc/ooni/prio.conf  Runs as a system daemon  Inputs:     Database tables:         `citizenlab` db table         country codes are in the `cc` column, lowercase, with the exception of ZZ  Outputs:     Files in /var/lib/analysis     Node exporter / prometheus metrics     Dedicated unlogged database tables and charts         tables:             currently_blocked  Special country code values:     ZZ: unknown     XX: all # debdeps: python3-systemd # @metrics.timer("update_url_prioritization") SELECT priority, domain, url, cc, category_code FROM citizenlab # Create dict: cc -> category_code -> [entry, ... ] # merge ZZ into each country Chao weighted random sampling # probability for this item # conf.refresh_interval: # category_code -> [test_item, ... ] https://orchestrate.ooni.io/api/v1/test-list/urls?country_code=IT # parsed values from DEFAULT section