<commit_before>#!/usr/bin/env python3
# -*- coding: utf-8 -*-

import os
import re
import sys
import math
import time
import json
import queue
import signal
import random
import logging
import sqlite3
import threading
import functools
import subprocess
import collections

import requests
from vendor import libirc

__version__ = '1.2'

MEDIA_TYPES = frozenset(('audio', 'document', 'photo', 'sticker', 'video', 'voice', 'contact', 'location', 'new_chat_participant', 'left_chat_participant', 'new_chat_title', 'new_chat_photo', 'delete_chat_photo', 'group_chat_created', '_ircuser'))

loglevel = logging.DEBUG if sys.argv[-1] == '-d' else logging.INFO

logging.basicConfig(stream=sys.stdout, format='# %(asctime)s [%(levelname)s] %(message)s', level=loglevel)

HSession = requests.Session()
USERAGENT = 'TgChatDiggerBot/%s %s' % (__version__, HSession.headers["User-Agent"])
HSession.headers["User-Agent"] = USERAGENT

db = sqlite3.connect('chatlog.db')
conn = db.cursor()
conn.execute('''CREATE TABLE IF NOT EXISTS messages (
id INTEGER PRIMARY KEY,
src INTEGER,
text TEXT,
media TEXT,
date INTEGER,
fwd_src INTEGER,
fwd_date INTEGER,
reply_id INTEGER
)''')
conn.execute('''CREATE TABLE IF NOT EXISTS users (
id INTEGER PRIMARY KEY,
username TEXT,
first_name TEXT,
last_name TEXT
)''')
conn.execute('CREATE TABLE IF NOT EXISTS config (id INTEGER PRIMARY KEY, val INTEGER)')
# conn.execute('CREATE TABLE IF NOT EXISTS words (word TEXT PRIMARY KEY, count INTEGER)')

re_ircaction = re.compile('^\x01ACTION (.*)\x01$')
re_ircforward = re.compile(r'^\[([^]]+)\] (.*)$|^\*\* ([^ ]+) (.*) \*\*$')

class LRUCache:

    def __init__(self, maxlen):
        self.capacity = maxlen
        self.cache = collections.OrderedDict()

    def __getitem__(self, key):
        value = self.cache.pop(key)
        self.cache[key] = value
        return value

    def get(self, key, default=None):
        try:
            value = self.cache.pop(key)
            self.cache[key] = value
            return value
        except KeyError:
            return default

    def __setitem__(self, key, value):
        try:
            self.cache.pop(key)
        except KeyError:
            if len(self.cache) >= self.capacity:
                self.cache.popitem(last=False)
        self.cache[key] = value

def async_func(func):
    @functools.wraps(func)
    def wrapped(*args, **kwargs):
        def func_noerr(*args, **kwargs):
            try:
                func(*args, **kwargs)
            except Exception:
                logging.exception('Async function failed.')
        thr = threading.Thread(target=func_noerr, args=args, kwargs=kwargs)
        thr.daemon = True
        thr.start()
    return wrapped

def _raise_ex(ex):
    raise ex

### Polling

def getupdates():
    global OFFSET, MSG_Q
    while 1:
        try:
            updates = bot_api('getUpdates', offset=OFFSET, timeout=10)
        except Exception as ex:
            logging.exception('Get updates failed.')
            continue
        if updates:
            logging.debug('Messages coming.')
            OFFSET = updates[-1]["update_id"] + 1
            for upd in updates:
                MSG_Q.put(upd)
        time.sleep(.2)

def checkappproc():
    global APP_P
    if APP_P.poll() is not None:
        APP_P = subprocess.Popen(APP_CMD, stdin=subprocess.PIPE, stdout=subprocess.PIPE)

def runapptask(cmd, args, sendargs):
    '''`sendargs` should be (chatid, replyid)'''
    global APP_P, APP_LCK, APP_TASK
    with APP_LCK:
        # Prevent float problems
        tid = str(time.time())
        text = json.dumps({"cmd": cmd, "args": args, "id": tid})
        APP_TASK[tid] = sendargs
        try:
            APP_P.stdin.write(text.strip().encode('utf-8') + b'\n')
            APP_P.stdin.flush()
        except BrokenPipeError:
            checkappproc()
            APP_P.stdin.write(text.strip().encode('utf-8') + b'\n')
            APP_P.stdin.flush()
        logging.debug('Wrote to APP_P: ' + text)

def getappresult():
    global APP_P, APP_TASK
    while 1:
        try:
            result = APP_P.stdout.readline().strip().decode('utf-8')
        except BrokenPipeError:
            checkappproc()
            result = APP_P.stdout.readline().strip().decode('utf-8')
        logging.debug('Got from APP_P: ' + result)
        if result:
            obj = json.loads(result)
            if obj['exc']:
                logging.error('Remote app server error.\n' + obj['exc'])
            sargs = APP_TASK.get(obj['id'])
            if sargs:
                sendmsg(obj['ret'] or 'Empty.', sargs[0], sargs[1])
                del APP_TASK[obj['id']]
            else:
                logging.error('Task ID %s not found.' % obj['id'])

def checkircconn():
    global ircconn
    if not ircconn or not ircconn.sock:
        ircconn = libirc.IRCConnection()
        ircconn.connect((CFG['ircserver'], CFG['ircport']), use_ssl=CFG['ircssl'])
        if CFG.get('ircpass'):
            ircconn.setpass(CFG['ircpass'])
        ircconn.setnick(CFG['ircnick'])
        ircconn.setuser(CFG['ircnick'], CFG['ircnick'])
        ircconn.join(CFG['ircchannel'])
        logging.info('IRC (re)connected.')

def getircupd():
    global MSG_Q, IRCOFFSET
    while 1:
        checkircconn()
        line = ircconn.parse(block=False)
        if line and line["cmd"] == "PRIVMSG":
            if line["dest"] != CFG['ircnick'] and not re.match(CFG['ircbanre'], line["nick"]):
                msg = {
                    'message_id': IRCOFFSET,
                    'from': {'id': CFG['ircbotid'], 'first_name': CFG['ircbotname'], 'username': 'orzirc_bot'},
                    'date': int(time.time()),
                    'chat': {'id': -CFG['groupid'], 'title': CFG['ircchannel']},
                    'text': line["msg"].strip(),
                    '_ircuser': line["nick"]<commit_msg>use blocking read from IRC<commit_after>#!/usr/bin/env python3
# -*- coding: utf-8 -*-

import os
import re
import sys
import math
import time
import json
import queue
import signal
import random
import logging
import sqlite3
import threading
import functools
import subprocess
import collections

import requests
from vendor import libirc

__version__ = '1.2'

MEDIA_TYPES = frozenset(('audio', 'document', 'photo', 'sticker', 'video', 'voice', 'contact', 'location', 'new_chat_participant', 'left_chat_participant', 'new_chat_title', 'new_chat_photo', 'delete_chat_photo', 'group_chat_created', '_ircuser'))

loglevel = logging.DEBUG if sys.argv[-1] == '-d' else logging.INFO

logging.basicConfig(stream=sys.stdout, format='# %(asctime)s [%(levelname)s] %(message)s', level=loglevel)

HSession = requests.Session()
USERAGENT = 'TgChatDiggerBot/%s %s' % (__version__, HSession.headers["User-Agent"])
HSession.headers["User-Agent"] = USERAGENT

db = sqlite3.connect('chatlog.db')
conn = db.cursor()
conn.execute('''CREATE TABLE IF NOT EXISTS messages (
id INTEGER PRIMARY KEY,
src INTEGER,
text TEXT,
media TEXT,
date INTEGER,
fwd_src INTEGER,
fwd_date INTEGER,
reply_id INTEGER
)''')
conn.execute('''CREATE TABLE IF NOT EXISTS users (
id INTEGER PRIMARY KEY,
username TEXT,
first_name TEXT,
last_name TEXT
)''')
conn.execute('CREATE TABLE IF NOT EXISTS config (id INTEGER PRIMARY KEY, val INTEGER)')
# conn.execute('CREATE TABLE IF NOT EXISTS words (word TEXT PRIMARY KEY, count INTEGER)')

re_ircaction = re.compile('^\x01ACTION (.*)\x01$')
re_ircforward = re.compile(r'^\[([^]]+)\] (.*)$|^\*\* ([^ ]+) (.*) \*\*$')

class LRUCache:

    def __init__(self, maxlen):
        self.capacity = maxlen
        self.cache = collections.OrderedDict()

    def __getitem__(self, key):
        value = self.cache.pop(key)
        self.cache[key] = value
        return value

    def get(self, key, default=None):
        try:
            value = self.cache.pop(key)
            self.cache[key] = value
            return value
        except KeyError:
            return default

    def __setitem__(self, key, value):
        try:
            self.cache.pop(key)
        except KeyError:
            if len(self.cache) >= self.capacity:
                self.cache.popitem(last=False)
        self.cache[key] = value

def async_func(func):
    @functools.wraps(func)
    def wrapped(*args, **kwargs):
        def func_noerr(*args, **kwargs):
            try:
                func(*args, **kwargs)
            except Exception:
                logging.exception('Async function failed.')
        thr = threading.Thread(target=func_noerr, args=args, kwargs=kwargs)
        thr.daemon = True
        thr.start()
    return wrapped

def _raise_ex(ex):
    raise ex

### Polling

def getupdates():
    global OFFSET, MSG_Q
    while 1:
        try:
            updates = bot_api('getUpdates', offset=OFFSET, timeout=10)
        except Exception as ex:
            logging.exception('Get updates failed.')
            continue
        if updates:
            logging.debug('Messages coming.')
            OFFSET = updates[-1]["update_id"] + 1
            for upd in updates:
                MSG_Q.put(upd)
        time.sleep(.2)

def checkappproc():
    global APP_P
    if APP_P.poll() is not None:
        APP_P = subprocess.Popen(APP_CMD, stdin=subprocess.PIPE, stdout=subprocess.PIPE)

def runapptask(cmd, args, sendargs):
    '''`sendargs` should be (chatid, replyid)'''
    global APP_P, APP_LCK, APP_TASK
    with APP_LCK:
        # Prevent float problems
        tid = str(time.time())
        text = json.dumps({"cmd": cmd, "args": args, "id": tid})
        APP_TASK[tid] = sendargs
        try:
            APP_P.stdin.write(text.strip().encode('utf-8') + b'\n')
            APP_P.stdin.flush()
        except BrokenPipeError:
            checkappproc()
            APP_P.stdin.write(text.strip().encode('utf-8') + b'\n')
            APP_P.stdin.flush()
        logging.debug('Wrote to APP_P: ' + text)

def getappresult():
    global APP_P, APP_TASK
    while 1:
        try:
            result = APP_P.stdout.readline().strip().decode('utf-8')
        except BrokenPipeError:
            checkappproc()
            result = APP_P.stdout.readline().strip().decode('utf-8')
        logging.debug('Got from APP_P: ' + result)
        if result:
            obj = json.loads(result)
            if obj['exc']:
                logging.error('Remote app server error.\n' + obj['exc'])
            sargs = APP_TASK.get(obj['id'])
            if sargs:
                sendmsg(obj['ret'] or 'Empty.', sargs[0], sargs[1])
                del APP_TASK[obj['id']]
            else:
                logging.error('Task ID %s not found.' % obj['id'])

def checkircconn():
    global ircconn
    if not ircconn or not ircconn.sock:
        ircconn = libirc.IRCConnection()
        ircconn.connect((CFG['ircserver'], CFG['ircport']), use_ssl=CFG['ircssl'])
        if CFG.get('ircpass'):
            ircconn.setpass(CFG['ircpass'])
        ircconn.setnick(CFG['ircnick'])
        ircconn.setuser(CFG['ircnick'], CFG['ircnick'])
        ircconn.join(CFG['ircchannel'])
        logging.info('IRC (re)connected.')

def getircupd():
    global MSG_Q, IRCOFFSET
    while 1:
        checkircconn()
        line = ircconn.parse(block=True)
        if line and line["cmd"] == "PRIVMSG":
            if line["dest"] != CFG['ircnick'] and not re.match(CFG['ircbanre'], line["nick"]):
                msg = {
                    'message_id': IRCOFFSET,
                    'from': {'id': CFG['ircbotid'], 'first_name': CFG['ircbotname'], 'username': 'orzirc_bot'},
                    'date': int(time.time()),
                    'chat': {'id': -CFG['groupid'], 'title': CFG['ircchannel']},
                    'text': line["msg"].strip(),
                    '_ircuser': line["nick"]
{'fraction_non_alphanumeric': 0.14466268146883007, 'fraction_numerical': 0.004269854824935952, 'mean_word_length': 2.5136513651365138, 'pattern_counts': {'":': 8, '<': 3, '<?xml version=': 0, '>': 5, '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': '3538883', 'n_tokens_mistral': 3918, 'n_tokens_neox': 3714, 'n_words': 873}