import os import glob import sys import docing import filer import dater from rixer import Rixer from staff import Staff from idref import Idref from spana import Spana # from xpath import Xpath import lxml.etree as et from xpafs import Xpafs class Recon: def __init__(self, erimp, do_verbose=False): self.e = erimp self.xpspaces = {'a': erimp.ns['amf'], 'e': erimp.ns['ernad']} self.do_verbose = do_verbose self.staff = Staff(self.e, do_verbose=do_verbose) self.rixer = Rixer(self.e, do_verbose=do_verbose) self.spana = Spana(self.e, do_verbose=do_verbose) self.xpafs = Xpafs(self.e, do_verbose=do_verbose) self.report = None self.do_verbose = do_verbose # # references that are already seen self.idref = Idref(self.e) # # more refs to do? self.we_have_more_refs = None # # individual configuration files, by id self.ficons = {} # # looese isssudates used for checking self.issuedates = None self.today = dater.today() ## # self.all_nouns = ('collection', 'person', 'organization') return None def get_oppoverb(self, verb): if verb == 'iseditorof': return 'haseditor' if verb == 'haseditor': return 'iseditorof' if verb == 'isreplacedby': return None # # should not exist # if verb == 'isforkedfrom': # return None return False def fill(self, what): e = self.e # folder = e.dirs['etc'] + '/reports/' + what folder = e.dirs['recon'] + '/' + what fufis = e.d.list_amf_fufis(folder, with_gz=False) cut_len = len(e.ext['amf']) recons = {} if fufis is None: print(f"recons sees no fufis in {folder}") quit() for fufi in fufis: bana = os.path.basename(fufi) repcode = bana[0:-cut_len] recons[repcode] = fufi return recons def amf_doc(self, repcode, add_state={}, do_refs=True): doc = self.doc(repcode, add_state=add_state, do_refs=do_refs) amf_doc = self.rixer.wrap_with_amf(doc.getroot()) et.cleanup_namespaces(amf_doc, top_nsmap=self.e.ns) return amf_doc def fufi(self, repcode): e = self.e fudi = e.dirs['etc'] + '/reports/available' glob_string = fudi + '/' + repcode + '*.xml' avail_fufis = glob.glob(glob_string) if len(avail_fufis) == 0: print(f"recon sees no file for {glob_string}", file=sys.stderr) return None if len(avail_fufis) > 1: print(f"recon sees several files for {glob_string}", file=sys.stderr) return None # doc = filer.parse_strict(avail_fufis[0]) return avail_fufis[0] def is_it_enabled(self, repcode): e = self.e fudi = e.dirs['etc'] + '/reports/enabled' glob_string = fudi + '/' + repcode + '*.xml' enab_fufis = glob.glob(glob_string) if len(enab_fufis) == 0: return False if len(enab_fufis) > 1: print(f"recon sees several files for {glob_string}", file=sys.stderr) return False # doc = filer.parse_strict(enab_fufis[0]) return True def bare_doc(self, repcode): fufi = self.fufi(repcode) if fufi is None: return None if not os.path.isfile(fufi): print(f"recon does not see {fufi}", file=sys.stderr) return None doc = filer.parse_lax(fufi) return doc def doc(self, repcode, add_state={}, do_refs=True, no_staff=False): """get the recon document""" ernad_ns = self.e.ns['ernad'] doc = self.bare_doc(repcode) doc_ele = doc.getroot() if not self.is_it_enabled(repcode): closed_ele = et.SubElement(doc_ele, et.QName(ernad_ns, 'closed')) closed_ele.text = 'closed' # # set if there is a latest link if self.latest_link(repcode): # latli_ele = et.SubElement(doc_ele, et.QName(ernad_ns, 'latli')) # closed_ele.text = 'closed' if no_staff: doc = self.rixer.decomment(doc) return doc # # add references to reports by the same staff if do_refs: # self.idref.clear() # # the stored docs don't have AMF wrapper # self.idref.note(doc.getroot(), 'id') self.we_have_more_refs = True doc = self.resolve_refs(doc) state_dir = self.e.report[repcode].dirs['state'] if self.e.conf['use_staff']: doc_ele = self.staff.inject(doc_ele, repcode) # # if a dict is give we use it as giving exceptions if do_refs: if isinstance(add_state, dict): glob_string = state_dir + '/*.xml' state_fufis = glob.glob(glob_string) for state_fufi in state_fufis: if self.do_verbose: print(f"recon adds {state_fufi}") state_bana = os.path.basename(state_fufi) # # remove .xml kind = state_bana[:-4] if kind in add_state: # # kind is an exception continue state_doc = filer.parse_lax(state_fufi) if state_doc is None: continue doc_ele.append(state_doc.getroot()) # # if a list is given, we add the elements of the list elif isinstance(add_state, list): for item in add_state: state_fufi = state_dir + '/' + item + '.xml' if not os.path.isfile(state_fufi): continue state_doc = filer.parse_lax(state_fufi) if state_doc is None: continue doc_ele.append(state_doc.getroot()) doc = self.rixer.decomment(doc) # if self.do_verbose: # print(docing.show(doc)) # # used to return False return doc def resolve_refs(self, doc, count=0): refs = self.rixer.get_refs(doc) if self.do_verbose: print('refs is ' + str(refs)) if len(refs) == 0: if self.do_verbose: print("no refs") return doc for ref in refs: if self.do_verbose: print('ref ' + ref) ref_ele = refs[ref] ref = ref_ele.attrib['ref'] noun = self.idref.noun(ref_ele) # print('is noted ' + str(self.idref.is_noted(ref_ele))) if noun in ('person', 'organization'): doc = self.staff.inject(doc, ref) continue resolved_doc = self.doc(ref, do_refs=False, add_state=None) # self.idref.note(ref_ele, 'ref') if resolved_doc is None: if self.do_verbose: print(f"staff can't resolve {ref}", file=sys.stderr) continue resolved_ele = resolved_doc.getroot() et.cleanup_namespaces(resolved_ele, top_nsmap=self.e.ns) # self.idref.note(ref_ele, 'ref') ref_ele.getparent().replace(ref_ele, resolved_ele) refs = self.rixer.get_refs(doc, level=count) self.idref.clear_refs_when_id(doc) self.idref.delete_by_subsequest_id(doc) if len(refs) == 0: return doc # # level does not seem to make a difference if count < 1: # print("start cont") self.resolve_refs(doc, count=count+1) return doc def get_current_editor(self, repcode): doc = self.doc(repcode) editor_xp = '/a:haseditor' editor_eles = doc.findall(editor_xp, namespaces=self.xpspaces) # # it does not the the not(@until) current_editors = [] for editor_ele in editor_eles: if 'until' in editor_ele.attrib: continue current_editors.append(editor_ele) editor_eles = current_editors total_editors = len(editor_eles) if total_editors != 1: print(f"recon: I can't deal with {total_editors} editors") return None return editor_eles[0] def all_docs(self, add_state={}, do_refs=True): """check relational data related to issues""" # folder = self.dirs['recon'] + '/available' # for fufi in glob.glob(f"{folder}/*") # # quit() docs = {} self.verbose = False for repcode in self.e.report: docs[repcode] = self.doc(repcode, add_state=add_state, do_refs=do_refs) return docs # # configuration parts in files def add_ficons(self, intype): fudi = self.e.dirs[intype] if not os.path.isdir(fudi): print("I don't see the {intype} directory", file=sys.stderr) return {} lenamf = len(self.e.ext['amf']) staff_glob = fudi + '/*' + self.e.ext['amf'] for fufi in glob.glob(staff_glob): handle = os.path.basename(fufi)[0:-lenamf] # # don't read this again if handle in self.ficons: continue self.ficons[handle] = filer.parse_lax(fufi) def check_dates(self): self.add_ficons('staff') self.add_ficons('avail') for handle in self.ficons: self.check_issuedates(handle) def check_issuedates(self, handle): doc = self.ficons[handle] # # self.isssudates are loose, self.e.issuedates are strict if self.issuedates is None: self.issuedates = self.e.folda.loose_dates(self.e.dirs['issues']) for dest in ('from', 'until'): xp = './/*[@' + dest + ']' # # this seems to have to be done through element found_eles = doc.findall(xp) for found_ele in found_eles: tag = found_ele.tag is_verb = self.idref.is_verb(tag) if not is_verb: continue date = found_ele.attrib[dest] if date not in self.issuedates: if date < self.today: print(docing.show(found_ele)) print(f"bad date {date} in config {handle}") quit() def mtime(self, repcode): # # some duplicate with "inject" method fufi = self.fufi(repcode) repdoc = self.doc(repcode, no_staff=True) max_mtime = os.path.getmtime(fufi) refs = self.rixer.get_refs(repdoc, only_verbs=['haseditor']) for ref in refs: # # in fact the ref is as stid fufi = self.staff.fufi(ref) if not os.path.isfile(fufi): continue mtime = os.path.getmtime(fufi) if mtime > max_mtime: max_mtime = mtime return max_mtime def remove_outdated(self, doc, issuedate): until_eles = doc.findall('.//*[@until]') for until_ele in until_eles: if issuedate > until_ele.attrib['until']: until_ele.getparent().remove(until_ele) # # Gabriele case else: del until_ele.attrib['until'] from_eles = doc.findall('.//*[@from]') for from_ele in from_eles: if issuedate < from_ele.attrib['from']: from_ele.getparent().remove(from_ele) return doc def barebones(self, doc, show_email=False): # # <-- ausfu for path in ('.//e:*', './/a:iseditorof', './/a:person/a:email'): if show_email and 'email' in path: continue path_eles = doc.findall(path, namespaces=self.xpspaces) for path_ele in path_eles: path_ele.getparent().remove(path_ele) return doc def latest_link(self, repcode): """has the web site link to the latest issue of the report?""" """in other words, has the report ever issued a paper?""" start_dir = os.getcwd() html_fudi = self.e.dirs['blatt'] + f'/reports/{repcode}' dates = self.e.folda.by_dates(html_fudi) if dates is None or len(dates) == 0: return False last_date = list(dates)[0] fufi = dates[last_date] bana = os.path.basename(fufi) fuli = html_fudi + '/latest.html' if os.path.islink(fuli): there_link = os.readlink(fuli) there_date = there_link[0:10] if there_date == last_date: if self.do_verbose: print(f"blata: link is there at {there_date}") return True os.remove(fuli) os.chdir(html_fudi) if self.do_verbose: print(f"blata links {bana} to latest.html in {html_fudi}") try: os.symlink(bana, 'latest.html') except FileExistsError: pass os.chdir(start_dir) return True def is_it_nopublish(self, doc): eles = doc.findall('//e:no_publish', namespaces=self.xpspaces) if len(eles) > 0: return True return False def united(self): root_ele = self.rixer.base('collection', 'amf') root_ele.attrib['id'] = self.e.impna all_docs = self.all_docs(add_state=[], do_refs=False) for repcode in all_docs: root_ele.append(all_docs[repcode].getroot()) return root_ele def check_rels(self): self.add_ficons('avail') self.add_ficons('staff') self.check_av2staff() self.check_bigwig() self.check_we_have_editor() def de_amf_ns(self, string): string = string[24:] return string def check_av2staff(self): amf_ns = '{' + self.e.ns['amf'] + '}' tags_to_ignore = (amf_ns + 'collection') for handle in self.ficons: if not handle.startswith(self.e.impna + '-'): continue doc = self.ficons[handle] target_eles = doc.xpath('//*[@ref]', namespaces=self.xpspaces) for target_ele in target_eles: verb_ele = target_ele.getparent() verb = self.de_amf_ns(verb_ele.tag) target_verb = self.get_oppoverb(verb) if target_verb is None: continue if target_verb is False: print(f"bad verb {verb} for {handle}, record garbled.") continue # # if there is child element, ignore this element tag = target_ele.tag if tag in tags_to_ignore: continue if len(target_ele.xpath('./*')) > 0: continue target_handle = target_ele.attrib['ref'] if target_handle not in self.ficons: print(f"No relations to {target_handle} seen in {handle}") continue target_doc = self.ficons[target_handle] xp = "//a:" + target_verb + "/*[@ref='" + handle + "']" target_nouns = target_doc.xpath(xp, namespaces=self.xpspaces) if len(target_nouns) == 0: mess = f"no {target_verb} {handle} in " mess += f"record {target_handle}\n" print(mess) continue if len(target_nouns) > 1: count = str(len(target_nouns)) mess = f"{count} {target_verb}s {handle} in record" mess += f"{target_handle}" print(mess) continue target_verb_ele = target_nouns[0].getparent() if self.spana.has_it_same(verb_ele, target_verb_ele): continue print("Incompatible spans") print(docing.show(verb_ele), end="") print(docing.show(target_verb_ele)) def check_bigwig(self): perid_repcode = {} bigwigs = [] for handle in self.ficons: if '-' not in handle: bigwigs.append(handle) for handle in self.ficons: if '-' not in handle: continue doc = self.ficons[handle] ed_eles = doc.xpath('/a:collection/a:haseditor/a:person', namespaces=self.xpspaces) for ed_ele in ed_eles: if 'ref' not in ed_ele.attrib: # print(f"recon: missing ref= in {handle}") continue perid = ed_ele.attrib['ref'] if perid in bigwigs: other_eles = ed_ele.xpath('./*', namespaces=self.xpspaces) if len(other_eles) > 0: print(f"{perid} in {handle}: bigwig and other data") continue if perid in perid_repcode: target = perid_repcode[perid] print(f"bigwig error {perid} of {handle} in {target}") perid_repcode[perid] = handle def check_person_handle_in_homepages(self): for handle in self.ficons: doc = self.ficons[handle] # # a primitive check to see if this is a person handle bigwigs = [] if '-' not in handle: self.check_person_handle_in_homepage_person(doc) bigwigs.append(handle) continue # print(handle + ' ', end="") self.check_person_handle_in_homepage_report(doc) def check_person_handle_in_homepage_person(self, doc): doc_string = docing.show(doc) ids = doc.xpath('/a:person/@id', namespaces=self.xpspaces) if len(ids) != 1: print(f"I need a handle in {doc_string}") return perid = ids[0].split('/')[-1] # print(doc_string) hopas = doc.xpath('/a:person/a:homepage', namespaces=self.xpspaces) if len(hopas) != 1: print(f"no hopa in {perid}") return main_hopa = hopas[0].text if perid not in main_hopa: print(f"recon: {perid} not in {main_hopa}") return def check_person_handle_in_homepage_report(self, doc): doc_string = docing.show(doc) editors = doc.xpath('/a:collection/a:haseditor/a:person', namespaces=self.xpspaces) if len(editors) == 0: # print(f"I need an editor in in {doc_string}") return for person_ele in editors: doc_string = docing.show(person_ele) # # do we have attributes? attribute_eles = person_ele.xpath('./*', namespaces=self.xpspaces) if len(attribute_eles) == 0: continue refs = person_ele.xpath('./@ref', namespaces=self.xpspaces) if len(refs) != 1: # print(f"I need a ref in {doc_string}") return peref = refs[0].split('/')[-1] xp = "./a:homepage" hopas = person_ele.xpath(xp, namespaces=self.xpspaces) if len(hopas) == 0: print(f"no hopa in {doc_string}") continue main_hopa = hopas[0].text if peref not in main_hopa: print(f"recon: {peref} not in {main_hopa}") print(f"{doc_string}") continue def check_we_have_editor(self): # amf_ns = '{' + self.e.ns['amf'] + '}' for handle in self.ficons: if not handle.startswith(self.e.impna + '-'): continue if not self.is_it_enabled(handle): continue doc = self.ficons[handle] editor_eles = doc.xpath('/a:collection/a:haseditor', namespaces=self.xpspaces) count_editors = 0 for editor_ele in editor_eles: if 'until' not in editor_ele.attrib: count_editors += 1 if count_editors == 0: doc_string = docing.show(doc) print(f"I need an editor in\n{doc_string}") def get_mothers(self, repcode): doc = self.doc(repcode) print(f"recon: start from {repcode}") # # remember that the doc resolves the ref= on the isedicevedfrom repcode_xp = '/a:collection/e:forked/@report' repcodes = self.xpafs.run(doc, repcode_xp) starts_xp = '/a:collection/e:forked/@start' starts = self.xpafs.run(doc, starts_xp) if len(repcodes) == 0: return None # if len(repcodes) > 1: # print("recon: enhance me, I can only do one parent") if len(starts) != len(repcodes): print("recon: bad forking data", file=sys.sterr) return None total_mothers = len(starts) mothers = {} count = 0 while count < total_mothers: mothers[repcodes[count]] = starts[count] count += 1 return mothers def jun_date(self, repcode): """first date that has actual papers""" html_fudi = self.e.dirs['blatt'] + f'/reports/{repcode}' dates = list(self.folder.by_dates(html_fudi).keys()) if len(dates) == 0: return None return dates[-1] def has_it_editor(self, repcode, issuedate): doc = self.bare_doc(repcode) buli = self.spana.has_it_rel(doc, 'haseditor', issuedate) return buli