buscmd.py

#!/usr/bin/env python
# -*- coding: utf-8 -*-

import click
import urllib.request, urllib.parse, urllib.error
import json
import os
import glob
import csv
import datetime
import io
import xlwt
import sys
import yaml
import shutil
import hashlib
import zlib
from pyparsing import Word, nums, alphanums, alphas, oneOf, lineStart, lineEnd, Optional, restOfLine, Literal, ParseException, CaselessLiteral
from urllib.request import URLopener

from pymongo import MongoClient, DESCENDING, ASCENDING

XMLTOJSON = '/usr/local/bin/xml-to-json'
#FILEPATH = '/home/_data/preservation/ftpmirrors/bus/bus.gov.ru/GeneralInfo'
LOCALPATH = '/data/ftp/bus.gov.ru/'
NSI_PATH = os.path.join(LOCALPATH, 'all')

DATA_CATEGORY_GENERAL = 'GeneralInfo'
DATA_CATEGORY_ACTIONGRANT = 'ActionGrant'
DATA_CATEGORY_ACTIVERES = 'ActivityResult'
DATA_CATEGORY_BUDGETCIRC = 'BudgetaryCircumstances'
DATA_CATEGORY_DIVERSEINFO = 'DiverseInfo'
DATA_CATEGORY_FINACTIVITY = 'FinancialActivityPlan'
DATA_CATEGORY_INSACTIVITY = 'InspectionActivity'
DATA_CATEGORY_STATETASK = 'StateTask'
DATA_CATEGORY_REGPUBINFO = 'regPublishedInfo'
DATA_CATEGORY_ABF0503121 = 'annualBalanceF0503121'
DATA_CATEGORY_ABF0503127 = 'annualBalanceF0503127'
DATA_CATEGORY_ABF0503130 = 'annualBalanceF0503130'
DATA_CATEGORY_ABF0503137 = 'annualBalanceF0503137'
DATA_CATEGORY_ABF0503721 = 'annualBalanceF0503721'
DATA_CATEGORY_ABF0503730 = 'annualBalanceF0503730'
DATA_CATEGORY_ABF0503737 = 'annualBalanceF0503737'

DATA_NSI_BUDGETS = 'nsiBudget'
DATA_NSI_KLADR = 'nsiKladr'
DATA_NSI_KBK = 'nsiKbk'
DATA_NSI_KBKBUDGET = 'nsiKbkBudget'
DATA_NSI_OKATO = 'nsiOkato'
DATA_NSI_OKTMO = 'nsiOktmo'
DATA_NSI_OKER = 'nsiOker'
DATA_NSI_OKVED = 'nsiOkved'
DATA_NSI_OKFS = 'nsiOkfs'
DATA_NSI_OKOGU = 'nsiOkogu'
DATA_NSI_OKOPF = 'nsiOkopf'
DATA_NSI_SUBJECTSERVICE = 'nsiSubjectService'
DATA_NSI_PPO = 'nsiPpo'
DATA_NSI_OGS = 'nsiOgs'
DATA_NSI_FEDERALSERVICE = 'nsiFederalService'
DATA_NSI_INSTITUTIONTYPE = 'nsiInstitutionType'
DATA_NSI_IT = 'nsiIt'

NSI_LIST = [DATA_NSI_BUDGETS, DATA_NSI_KLADR, DATA_NSI_KBK, DATA_NSI_KBKBUDGET, DATA_NSI_OKATO, 
    DATA_NSI_OKTMO, DATA_NSI_OKER, DATA_NSI_OKVED, DATA_NSI_OKFS,
    DATA_NSI_OKOGU, DATA_NSI_OKOPF, DATA_NSI_SUBJECTSERVICE, DATA_NSI_PPO, DATA_NSI_OGS, DATA_NSI_FEDERALSERVICE,
    DATA_NSI_IT, DATA_NSI_INSTITUTIONTYPE
]

CAT_MAP = {
    DATA_CATEGORY_GENERAL : 'generalinfo',
    DATA_CATEGORY_ACTIONGRANT : 'actiongrant',
    DATA_CATEGORY_ACTIVERES : 'activeres',
    DATA_CATEGORY_BUDGETCIRC : 'budgetcirc',
    DATA_CATEGORY_DIVERSEINFO : 'diverseinfo',
    DATA_CATEGORY_FINACTIVITY : 'finactplan',
    DATA_CATEGORY_INSACTIVITY : 'insactivity',
    DATA_CATEGORY_STATETASK : 'statetask',
    DATA_CATEGORY_REGPUBINFO : 'regpubinfo',
    DATA_CATEGORY_ABF0503121 : 'afb0503121',
    DATA_CATEGORY_ABF0503127 : 'afb0503127',
    DATA_CATEGORY_ABF0503130 : 'afb0503130',
    DATA_CATEGORY_ABF0503137 : 'afb0503127',
    DATA_CATEGORY_ABF0503721 : 'afb0503721',
    DATA_CATEGORY_ABF0503730 : 'afb0503730',
    DATA_CATEGORY_ABF0503737 : 'afb0503737',
}

ARCHIVE_STATUS_UNPROCESSED = 1
ARCHIVE_STATUS_PROCESSING = 2
ARCHIVE_STATUS_PROCESSED = 3


TMPPATH = '/tmp/ram'

def calc_hash_crc(filename):
    """Calculate hash and crc32 of selected file"""
    data = open(filename, 'rb').read()
    fhash = hashlib.sha256(data).hexdigest()
    fcrc = zlib.crc32(data)
    return {'sha256': fhash, 'crc32' : fcrc}

class MongoLoader:
    def __init__(self, thepath):
        self.localpath = thepath
        self.conn = MongoClient()
        self.db = self.conn['bus']
        self.archcoll = self.db['archives']
        self.filecoll = self.db['files']
        pass

    def buildIndexes(self):
        self.archcoll.ensure_index([('obj_name', DESCENDING)])
        for name in list(CAT_MAP.values()):
            self.db[name].ensure_index([('position.positionId', DESCENDING)])
        for name in NSI_LIST:
            self.db[name].ensure_index([('position.positionId', DESCENDING)])

    def stats(self):
        """Calc collections stats"""
        import tabulate
        keys = ['category', 'total_files', 'not_loaded', 'count', 'size']
        table = []
        print('Collecting stats')
        cats = list(CAT_MAP.keys())
        cats.sort()
        total = ['total', 0, 0, 0, 0]
        for cat in cats:
            row = [cat]
            row.append(self.archcoll.find({'category' : cat}).count())
            row.append(self.archcoll.find({'category' : cat, 'status' : ARCHIVE_STATUS_UNPROCESSED}).count())
            stats = self.db.command("collstats", CAT_MAP[cat])
            row.append(stats['count'])
            row.append(float(stats['size']) / (1024*1024))
            table.append(row)

        NSI_LIST.sort()
        for cat in NSI_LIST:
            row = [cat]
            row.append(self.archcoll.find({'category' : cat}).count())
            row.append(self.archcoll.find({'category' : cat, 'status' : ARCHIVE_STATUS_UNPROCESSED}).count())
            stats = self.db.command("collstats", cat)
            row.append(stats['count'])
            row.append(float(stats['size']) / (1024*1024))
            table.append(row)
        
        for row in table:            
            total[1] += row[1]
            total[2] += row[2]
            total[3] += row[3]
            total[4] += row[4]
        table.append(total) 
        print(tabulate.tabulate(table, headers=keys))


    def loadNSI(self):
        """Loads all NSI to DB"""
        for nsiname in NSI_LIST:
            print(nsiname)
#            if nsiname != DATA_NSI_KBKBUDGET: continue
            ffilter = NSI_PATH + '/' + nsiname + '_*.zip'
#            print ffilter
            files = glob.glob(ffilter)
            for pname in files:
                obj_name = pname.rsplit('/', 1)[1]
                obj = self.archcoll.find_one({'obj_name' : obj_name})
                if not obj:
                    obj = {'obj_name' : obj_name, 'filepath' : pname, 'category' : nsiname, 'filesize' : os.stat(pname)[6], 'status' : ARCHIVE_STATUS_UNPROCESSED}
                    obj.update(calc_hash_crc(pname))
                    self.archcoll.save(obj)
                    print('-', pname, ':', obj_name, 'saved')
                else:
                    print('-', pname, 'already collected')
            pass
        pass            


    def findArchives(self, category=DATA_CATEGORY_GENERAL, dbupdate=True):
        """Collect all archives by category"""
        print('Collecting source data from', self.localpath)
        pathname = os.path.join(self.localpath, category)
#        cat_dirs = os.listdir(self.localpath)
        for dirname, dirnames, filenames in os.walk(pathname):
            for filename in filenames:
                pname = os.path.join(dirname, filename)
                ext = pname.rsplit('.', 1)[-1].lower()
                if ext == 'zip':
                    region = pname.split('/')[-2]
                    obj_name = '/'.join(pname.split('/')[-3:])
                    obj = self.archcoll.find_one({'obj_name' : obj_name})
                    if not obj:
                        obj = {'obj_name' : obj_name, 'filepath' : pname, 'category' : category, 'region' : region, 'filesize' : os.stat(pname)[6], 'status' : ARCHIVE_STATUS_UNPROCESSED}
                        obj.update(calc_hash_crc(pname))
                        self.archcoll.save(obj)
                        print('-', pname, ':', obj_name, 'saved')
                    else:
                        print('-', pname, 'already collected')
#                        print obj

    def process_category(self, category, force=False, nocheck=False):
        """Process selected category"""
        all = []
        if not force:
            objects = self.archcoll.find({'category' : category, 'status' : {'$in' : [ARCHIVE_STATUS_UNPROCESSED, ARCHIVE_STATUS_PROCESSING]}})
            print('Total unprocessed archives:', objects.count())
        else:
            objects = self.archcoll.find({'category' : category})
            print('Total archives', objects.count())
        for o in objects:
            all.append(o)
        for o in all:
            self.process_archive(o, force=force, nocheck=nocheck)
        pass

    def process_archive(self, o, force=False, nocheck=False):
        if o['status'] == ARCHIVE_STATUS_UNPROCESSED:
            nocheck = True
        elif o['status'] == ARCHIVE_STATUS_PROCESSING:
            nocheck = False
        pname = o['filepath']
        name = pname.rsplit('/', 1)[-1]
        finndame = os.path.join(TMPPATH, name)
        try:
            os.mkdir(finndame)
        except:
            pass
        o['status'] = ARCHIVE_STATUS_PROCESSING
        tempname = os.path.join(finndame, name)
        shutil.copy(pname, tempname)
        cwd = os.getcwd()
        os.chdir(finndame)
        os.system('7za x -y %s 1> /dev/null 2>/dev/null' % (name))
        files = os.listdir(finndame)
        allfiles = []
        for fname in files:
            allfiles.append({'fname' : fname})
            self.process_file(o['category'], o['region'] if 'region' in o else None, name, os.path.abspath(fname), nocheck=nocheck, force=force)
        o['status'] = ARCHIVE_STATUS_PROCESSED
        o['files'] = allfiles
        o['nfiles'] = len(files)
        self.archcoll.save(o)
        os.chdir(cwd)
        shutil.rmtree(finndame)
        print(pname, 'processed')
        pass

    def process_file(self, category, region, archivename, filename, nocheck=False, force=True):
        collname = self.map_category(category)
        coll = self.db[collname]
        name, ext = filename.rsplit('.', 1)
        if ext == 'xml':
            if region:
                id = '/'.join([category, region, archivename, filename.rsplit('/', 1)[-1]])
            else:
                id = '/'.join([category, archivename, filename.rsplit('/', 1)[-1]])
                nocheck = True 
            if nocheck:           
                data = None 
            else:
                data = coll.find_one({'processed.id' : id})
            found = data is not None
            if data and not force:
                return

            os.system(XMLTOJSON + ' -t ns2:position %s > %s' % (filename, name + '.json'))
            f = file(name + '.json', 'r')
            objects = []
            if region:
                try:
                    fdata = json.loads(f.read())
                    objects.append(fdata)
                except ValueError:
                    return
            else:
                for r in f:
                    fdata = json.loads(r)
                    objects.append(fdata)            
            if len(objects) == 1 and not nocheck:
                if data is None: data = objects[0]
                else: data.update(objects[0])
                proc = {'id': id, 'category' : category, 'region' : region, 'xml' : filename, 'archive': archivename}
                data['processed'] = proc
                coll.save(data)
            else:
                for o in objects:
                    proc = {'id': id, 'category' : category, 'region' : region, 'xml' : filename, 'archive': archivename}
                    o['processed'] = proc
                    coll.save(o)
            if found:
                print('-', name, 'updated')
            else:
                print('-', name, 'saved')
        pass

    def map_category(self, category):
        """Returns collection name on category"""
        v = CAT_MAP.get(category, None)
        if v is None:
            if category in NSI_LIST:
                return category
            else:
                return None
        else:
            return v

    def reset(self):
        """Drops databases"""
        print('Reset database')
        self.archcoll.drop()
        print('archive collection dropped')
        for name in list(CAT_MAP.values()):
            print(name, 'collection dropped')
            self.db[name].drop()


@click.group()
def cli1():
    pass

@cli1.command()
@click.option('--cat', default=DATA_CATEGORY_GENERAL)
def listdata(cat):
    """Lists all source files"""
    loader = MongoLoader(LOCALPATH)
    loader.buildIndexes()
    loader.findArchives(cat)
    pass

@click.group()
def cli2():
    pass

@cli2.command()
@click.option('--cat', default=DATA_CATEGORY_GENERAL)
@click.option('--force', default=False)
@click.option('--nocheck', default=False)
def process(cat, force, nocheck):
    """Process archive files"""
    force = bool(force)
    nocheck = bool(nocheck)
    loader = MongoLoader(LOCALPATH)
    #loader.findArchives(DATA_CATEGORY_GENERAL)
    loader.buildIndexes()
    loader.process_category(cat, nocheck=nocheck, force=force)
    pass


@click.group()
def cli3():
    pass

@cli3.command()
def reset():
    """Drop databases"""
    loader = MongoLoader(LOCALPATH)
    loader.reset()


@click.group()
def cli4():
    pass

@cli4.command()
def stats():
    """Statistics"""
    loader = MongoLoader(LOCALPATH)
    loader.buildIndexes()
    loader.stats()
    pass

@click.group()
def cli5():
    pass

@cli5.command()
def loadnsi():
    """Load all NSI data"""
    loader = MongoLoader(LOCALPATH)
    loader.loadNSI()
    pass


cli = click.CommandCollection(sources=[cli1, cli2, cli3, cli4, cli5])

if __name__ == '__main__':
    cli()