mirror of
https://github.com/NNTmux/newznab-tmux.git
synced 2026-08-29 17:28:55 +00:00
Darko
191 lines
7.8 KiB
Python
191 lines
7.8 KiB
Python
#!/usr/bin/env python
|
|
# -*- coding: utf-8 -*-
|
|
|
|
import sys, os, time
|
|
import threading
|
|
try:
|
|
import queue
|
|
except ImportError:
|
|
import Queue as queue
|
|
import subprocess
|
|
import string
|
|
import signal
|
|
import datetime
|
|
import math
|
|
|
|
import lib.info as info
|
|
from lib.info import bcolors
|
|
conf = info.readConfig()
|
|
cur = info.connect()
|
|
start_time = time.time()
|
|
pathname = os.path.abspath(os.path.dirname(sys.argv[0]))
|
|
|
|
print(bcolors.HEADER + "\nBackfill Safe Threaded Started at {}".format(datetime.datetime.now().strftime("%H:%M:%S")) + bcolors.ENDC)
|
|
|
|
cur[0].execute("SELECT g.name FROM groups g LEFT JOIN shortgroups ON shortgroups.name = g.name WHERE shortgroups.name IS NULL AND backfill = 1")
|
|
dorun = cur[0].fetchone()
|
|
|
|
#close connection to mysql
|
|
info.disconnect(cur[0], cur[1])
|
|
|
|
if dorun:
|
|
#before we get the groups, lets update shortgroups
|
|
subprocess.call(["php", pathname+"/../bin/update_groups.php", ""])
|
|
else:
|
|
cur = info.connect()
|
|
cur[0].execute("SELECT name FROM shortgroups")
|
|
dorun = cur[0].fetchone()
|
|
info.disconnect(cur[0], cur[1])
|
|
if len(sys.argv) > 1 and sys.argv[1] not in dorun:
|
|
#before we get the groups, lets update shortgroups
|
|
subprocess.call(["php", pathname+"/../bin/update_groups.php", ""])
|
|
|
|
count = 0
|
|
previous = "'alt.binaries.crap'"
|
|
|
|
#if the group has less than 10000 to grab, just grab them, and loop another group
|
|
while count < 10000:
|
|
#get values from db
|
|
cur = info.connect()
|
|
cur[0].execute("SELECT (SELECT value FROM site WHERE setting = 'backfillthreads') AS a, (SELECT value FROM tmux WHERE setting = 'backfill_qty') AS b, (SELECT value FROM tmux WHERE setting = 'backfill') AS c, (SELECT value FROM tmux WHERE setting = 'backfill_order') AS e, (SELECT value FROM tmux WHERE setting = 'backfill_days') AS f, (SELECT value FROM site WHERE setting = 'maxmssgs') AS g")
|
|
dbgrab = cur[0].fetchall()
|
|
run_threads = int(dbgrab[0][0])
|
|
backfill_qty = int(dbgrab[0][1])
|
|
type = int(dbgrab[0][2])
|
|
intorder = int(dbgrab[0][3])
|
|
intbackfilltype = int(dbgrab[0][4])
|
|
maxmssgs = int(dbgrab[0][5])
|
|
|
|
#get the correct oder by for the query
|
|
if intorder == 1:
|
|
group = "ORDER BY first_record_postdate DESC"
|
|
elif intorder == 2:
|
|
group = "ORDER BY first_record_postdate ASC"
|
|
elif intorder == 3:
|
|
group = "ORDER BY name ASC"
|
|
elif intorder == 4:
|
|
group = "ORDER BY name DESC"
|
|
elif intorder == 5:
|
|
group = "ORDER BY a.last_record DESC"
|
|
else:
|
|
group = "ORDER BY a.last_record ASC"
|
|
|
|
#backfill days or safe backfill date
|
|
if intbackfilltype == 1:
|
|
backfilldays = "backfill_target"
|
|
elif intbackfilltype == 2:
|
|
backfilldays = "datediff(curdate(),(SELECT value FROM site WHERE setting = 'safebackfilldate'))"
|
|
|
|
#query to grab backfill groups
|
|
if len(sys.argv) == 1:
|
|
if conf['DB_TYPE'] == "mysql":
|
|
cur[0].execute("SELECT g.name, g.first_record AS our_first, MAX(a.first_record) AS thier_first, MAX(a.last_record) AS their_last FROM groups g INNER JOIN shortgroups a ON g.name = a.name WHERE g.first_record IS NOT NULL AND g.first_record_postdate IS NOT NULL AND g.backfill = 1 AND (NOW() - INTERVAL %s DAY) < g.first_record_postdate AND g.name NOT IN (%s) GROUP BY a.name, a.last_record, g.name, g.first_record %s LIMIT 1" % (backfilldays, previous, group))
|
|
elif conf['DB_TYPE'] == "pgsql":
|
|
cur[0].execute("SELECT g.name, g.first_record AS our_first, MAX(a.first_record) AS thier_first, MAX(a.last_record) AS their_last FROM groups g INNER JOIN shortgroups a ON g.name = a.name WHERE g.first_record IS NOT NULL AND g.first_record_postdate IS NOT NULL AND g.backfill = 1 AND (NOW() - INTERVAL '%s DAYS') < g.first_record_postdate GROUP BY a.name, a.last_record, g.name, g.first_record %s LIMIT 1" % (backfilldays, group, groups))
|
|
datas = cur[0].fetchone()
|
|
else:
|
|
run = "SELECT g.name, g.first_record AS our_first, MAX(a.first_record) AS thier_first, MAX(a.last_record) AS their_last FROM groups g INNER JOIN shortgroups a ON g.name = a.name WHERE g.name = %s AND g.first_record IS NOT NULL AND g.first_record_postdate IS NOT NULL AND g.backfill = 1 LIMIT 1"
|
|
cur[0].execute(run, (sys.argv[1]))
|
|
datas = cur[0].fetchone()
|
|
if not datas or datas[0] is None:
|
|
print(bcolors.ERROR + "No Groups enabled for backfill" + bcolors.ENDC)
|
|
info.disconnect(cur[0], cur[1])
|
|
sys.exit()
|
|
|
|
#close connection to mysql
|
|
info.disconnect(cur[0], cur[1])
|
|
|
|
previous += ", '%s'" % datas[0]
|
|
count = datas[1] - datas[2]
|
|
if count < 0:
|
|
print(bcolors.ERROR + "USP returned an invalid first_post for {}, skipping it.".format(datas[0]) + bcolors.ENDC)
|
|
if len(sys.argv) == 2:
|
|
sys.exit()
|
|
|
|
if count == 0:
|
|
if len(sys.argv) == 2:
|
|
print(bcolors.ERROR + "We have hit the maximum we can backfill for {}, disabling it".format(datas[0]) + bcolors.ENDC)
|
|
remove = "UPDATE groups SET backfill = 0 WHERE name = %s"
|
|
cur = info.connect()
|
|
cur[0].execute(remove, (sys.argv[1]))
|
|
cur[1].autocommit(True)
|
|
info.disconnect(cur[0], cur[1])
|
|
sys.exit()
|
|
else:
|
|
print(bcolors.ERROR + "We have hit the maximum we can backfill for {}, skipping it".format(datas[0]) + bcolors.ENDC)
|
|
|
|
if count < 10000 and count > 0:
|
|
print(bcolors.PRIMARY + "Group {} has {} articles, in the range {} to {}".format(datas[0], "{:,}".format(count), "{:,}".format(datas[2]), "{:,}".format(datas[3])) + bcolors.ENDC)
|
|
print(bcolors.PRIMARY + "Our oldest post is: {}".format("{:,}".format(datas[1])) + bcolors.ENDC)
|
|
print(bcolors.PRIMARY + "Available Posts: {}".format("{:,}".format(count)) + bcolors.ENDC)
|
|
group = ("{} {}".format(datas[0], count))
|
|
subprocess.call(["php", pathname+"/../../multiprocessing/.do_not_run/switch.php", "python backfill_all_quantity "+str(group)])
|
|
|
|
#calculate the number of items for queue
|
|
if (count > (backfill_qty * run_threads)):
|
|
geteach = math.ceil((backfill_qty * run_threads) / maxmssgs)
|
|
else:
|
|
geteach = int(count / maxmssgs)
|
|
|
|
my_queue = queue.Queue()
|
|
time_of_last_run = time.time()
|
|
|
|
class queue_runner(threading.Thread):
|
|
def __init__(self, my_queue):
|
|
threading.Thread.__init__(self)
|
|
self.my_queue = my_queue
|
|
|
|
def run(self):
|
|
global time_of_last_run
|
|
|
|
while True:
|
|
try:
|
|
my_id = self.my_queue.get(True, 1)
|
|
except:
|
|
if time.time() - time_of_last_run > 3:
|
|
return
|
|
else:
|
|
if my_id:
|
|
time_of_last_run = time.time()
|
|
subprocess.call(["php", pathname+"/../../multiprocessing/.do_not_run/switch.php", "python "+my_id])
|
|
time.sleep(.03)
|
|
self.my_queue.task_done()
|
|
|
|
def main(args):
|
|
global time_of_last_run
|
|
time_of_last_run = time.time()
|
|
|
|
print(bcolors.HEADER + "We will be using a max of {} threads, a queue of {} and grabbing {} headers".format(run_threads, "{:,}".format(geteach), "{:,}".format(geteach * maxmssgs)) + bcolors.ENDC)
|
|
time.sleep(2)
|
|
|
|
def signal_handler(signal, frame):
|
|
sys.exit(0)
|
|
signal.signal(signal.SIGINT, signal_handler)
|
|
|
|
if True:
|
|
#spawn a pool of place worker threads
|
|
for i in range(run_threads):
|
|
p = queue_runner(my_queue)
|
|
p.setDaemon(False)
|
|
p.start()
|
|
|
|
#now load some arbitrary jobs into the queue
|
|
for i in range(0, int(geteach)):
|
|
time.sleep(.03)
|
|
my_queue.put("get_range backfill %s %s %s %s" % (datas[0], datas[1] - i * maxmssgs - maxmssgs, datas[1] - i * maxmssgs - 1, i+1))
|
|
|
|
my_queue.join()
|
|
|
|
group = ("{} {}".format(datas[0], 1000))
|
|
subprocess.call(["php", pathname+"/../../multiprocessing/.do_not_run/switch.php", "python backfill_all_quantity "+str(group)])
|
|
if run_threads <= geteach:
|
|
print(bcolors.HEADER + "\nWe used {} threads, a queue of {} and grabbed {} headers".format(run_threads, "{:,}".format(geteach), "{:,}".format(geteach * maxmssgs)) + bcolors.ENDC)
|
|
else:
|
|
print(bcolors.HEADER + "\nWe used {} threads, a queue of {} and grabbed {} headers".format(geteach, "{:,}".format(geteach), "{:,}".format(geteach * maxmssgs)) + bcolors.ENDC)
|
|
|
|
print(bcolors.HEADER + "\nBackfill Safe Threaded Completed at {}".format(datetime.datetime.now().strftime("%H:%M:%S")) + bcolors.ENDC)
|
|
print(bcolors.HEADER + "Running time: {}\n\n".format(str(datetime.timedelta(seconds=time.time() - start_time))) + bcolors.ENDC)
|
|
|
|
|
|
if __name__ == '__main__':
|
|
main(sys.argv[1:]) |