사용자:ego(비교)

api_url, wiki_url, discuss_doc, discuss_doc = selctwiki()

headers = {'Content-Type': 'application/json', 'Authorization': 'Bearer {}'.format(APItoken)}

discuss_headers = {'User-Agent': 'Mozilla/5.0'}

####

sys_stop=False

logger = logging.getLogger("wiki edit")

discusslogger = logging.getLogger("discuss HTML")

streamlogger = logging.getLogger("log stream")

logging.addLevelName(11, "edit History")

logging.editHistory = 11

logging.addLevelName(31, "edit Fail")

logging.editFail= 31

logging.addLevelName(41, "edit Error")

logging.editError = 41

logging.addLevelName(35, "working Emergency Stop")

logging.workingEmergencyStop = 35

logging.addLevelName(25, "discuss Error")

logging.discussError = 25

logging.addLevelName(12, "discuss Load")

logging.discussLoad = 12

logger.setLevel(logging.DEBUG)

discusslogger.setLevel(logging.INFO)

formatter = logging.Formatter('%(asctime)s - %(levelname)s - %(message)s')

stream_hander = logging.StreamHandler()

stream_hander.setFormatter(formatter)

streamlogger.addHandler(stream_hander)

progStart_day=str(time.strftime('%Y %m %d', time.localtime()))

progStart_time=str(time.strftime('%H %M %S', time.localtime()))

if not os.path.exists('/content/drive/My Drive/Colab Notebooks/logs/{}'.format(progStart_day)):

os.makedirs('/content/drive/My Drive/Colab Notebooks/logs/{}'.format(progStart_day))

os.makedirs('/content/drive/My Drive/Colab Notebooks/logs/{}/{}'.format(progStart_day,progStart_time))

file_handler = logging.FileHandler('/content/drive/My Drive/Colab Notebooks/logs/{}/{}/wiki edit.log'.format(progStart_day,progStart_time))

file_handler.setFormatter(formatter)

logger.addHandler(file_handler)

discuss_file_handler = logging.FileHandler('/content/drive/My Drive/Colab Notebooks/logs/{}/{}/discuss HTML.log'.format(progStart_day,progStart_time))

discuss_file_handler.setFormatter(formatter)

discusslogger.addHandler(discuss_file_handler)

error_file_handler = logging.FileHandler('/content/drive/My Drive/Colab Notebooks/logs/{}/{}/error.log'.format(progStart_day,progStart_time))

error_file_handler.setFormatter(formatter)

streamlogger.addHandler(error_file_handler)

class checkWikiThread():

async def alterResponseTry(self,response,wait_time=0.2):

try:

await response.html.arender(wait=wait_time)

html = await response.html.html

loadComple = re.search("토론",str(html))

if loadComple==None:

logger.log(logging.discussLoad, ('not Loding, wait_time: {}'.format(wait_time)))

print('not Loding, wait_time:',wait_time)

if wait_time>40:

discusslogger.log(logging.discussError,(html))

logger.log(logging.discussError, ("timeout, html log saved: {} discuss HTML.log".format(str(progStart_time))))

streamlogger.log(logging.discussError, ("timeout, html log saved: {} discuss HTML.log".format(str(progStart_time))))

html= await self.alterResponseTry(response,wait_time=float(wait_time+10))

else :

logger.log(logging.discussLoad, 'load comple')

return html

except Exception as ex:

logger.log(logging.discussError, ex)

streamlogger.log(logging.discussError, ex)

html= await self.alterResponseTry(response,wait_time)

return html

def __init__(self):

self.state=False

self.discuss_state = None

async def run(self):

self.state=True

response = await requests.get("https://{}/discuss/{}".format(wiki_url,discuss_doc),headers=discuss_headers, timeout=100)

html = response.text

loadComple = re.search("토론",str(html))

if loadComple==None:

logger.log(logging.discussLoad, 'not Loding, alter way start')

#print(html)

asession = await AsyncHTMLSession()

response = await asession.get("https://{}/discuss/{}".format(wiki_url,discuss_doc),headers=discuss_headers)

html= await self.alterResponseTry(response)

self.discuss_state = re.search("\/discuss\/[\w,%,:]*#s-1",str(html))

time.sleep(30)

self.state = False

def linkcheckchange(match_obj):

#구버전 기본값

#print(match_obj)

#print(match_obj[0])

#link = match_obj.group('link')

#anchor = match_obj.group('anchor')

#content = match_obj.group('content')

#namebar = match_obj.group('namebar')

#for target,replace in targets:

#if link==target:

#if content==None:

#내용이 없을경우 원래 링크가 내용이 됨

#content=link

#link=replace

#if anchor==None:

#anchor=''

#if namebar==None:

#namebar=''

#return '[[{}{}{}|{}]]'.format(namebar,link,anchor,content)

#

#[[]] 전용

first = match_obj.group('first')

link = match_obj.group('link')

middle = match_obj.group('middle')

detachBar = match_obj.group('detachBar')

content = match_obj.group('content')

last = match_obj.group('last')

anchor = match_obj.group('anchor')

if middle==None:

middle=''

for target in targets:

target_doc=target['document']

replace=target['replace']

if link==target_doc:

#내용이 없을경우 원래 링크가 내용이 됨

#content_replace bool 추가

if content==None and content_replace:

#혹시 모를 사고 예방

if detachBar==None:

content='|{}'.format(link)

else:

content='{}'.format(link)

elif content==None:

content=''

link=replace

return '{}{}{}{}{}'.format(first,link,middle,content,last)

return match_obj.group(0)

def checkchange(match_obj):

#[[]] 제외 기본 변경

first = match_obj.group('first')

link = match_obj.group('link')

last = match_obj.group('last')

for target in targets:

target_doc=target['document']

replace=target['replace']

if link==target_doc:

link=replace

#print('hidden')

return '{}{}{}'.format(first,link,last)

return match_obj.group(0)

def checkdelete(match_obj):

#분류 정리 전용

link = match_obj.group('link')

anchor = match_obj.group('anchor')

content = match_obj.group('content')

for target in targets:

target_doc=target['document']

replace=target['replace']

if link==target_doc:

if content==None:

#내용이 없을경우 원래 링크가 내용이 됨

content=link

link=replace

if anchor==None:

anchor=''

return ''

return match_obj.group(0)

def edit_link(title, text):

link_patten=re.compile(r"(?P<first>\[\[(?P<namebar>(:)?))(?P<link>((\\\\)|(\\\#)|(\\\|)|(\\\[)|(\\\])|[^\[\]#|])*)(?P<middle>(?P<anchor>#((\\\\)|(\\\#)|(\\\|)|(\\\[)|(\\\])|[^\[\]#|])*)?(?P<detachBar>\|)?)(?(detachBar)(?P<content>((\\\\)|(\\\#)|(\\\|)|(\\\[)|(\\\])|[^\[\]|])*))(?P<last>\]\])")

otherMean_patten=re.compile(r"(?P<first>rd\d=)(?P<link>((\\\,)|(\\\\)|(\\\))|[^\,)])*)(?P<last>[,)])")

detailContent_patten=re.compile(r"(?P<first>문서명=)(?P<link>((\\\,)|(\\\\)|(\\\))|[^\,)])*)(?P<last>[,)])")

categoryMean_patten=re.compile(r"(?P<first>n\d=)(?P<link>((\\\,)|(\\\\)|(\\\))|[^\,)])*)(?P<last>[,)])")

top_patten=re.compile(r"(?P<first>top\d=)(?P<link>((\\\,)|(\\\\)|(\\\))|[^\,)])*)(?P<last>[,)])")

include_patten=re.compile(r"(?P<first>\[include\()(?P<link>((\\\,)|(\\\\)|(\\\))|[^\,)])*)(?P<last>((?P<comma>,)?)?(?(comma)((\\\,)|(\\\\)|(\\\))|[^)])*)\)\])")

redirect_patten=re.compile(r"(?P<first>#redirect )(?P<link>((\\\\)|(\\\#)|(\\\|)|(\\\[)|(\\\])|[^\[\]#|])*)(?P<last>(?P<anchor>#)?.*\n)")

normal_new_text=link_patten.sub(linkcheckchange,text)

new_text=otherMean_patten.sub(checkchange,normal_new_text)

new_text=detailContent_patten.sub(checkchange,new_text)

new_text=categoryMean_patten.sub(checkchange,new_text)

new_text=top_patten.sub(checkchange,new_text)

new_text=include_patten.sub(checkchange,new_text)

new_text=redirect_patten.sub(checkchange,new_text)

if normal_new_text!=new_text:

logger.log(logging.editHistory,('특수 변경 문서명: {}'.format(title)))

return new_text

def backlinkchange():

data={}

data['edit']=[]

for target in targets:

target_doc=target['document']

replace=target['replace']

target_backlinks = []

backlink_value = ''

namespaces=[]

res = requests.get('https://{}/api/backlink/{}'.format(api_url,target_doc), headers = headers).json()

for namespace in res['namespaces']:

namespaces.append(namespace['namespace'])

if len(black_namespaces)>0: #블랙 네임이 있다면

for black_namespace in black_namespaces:

if black_namespace in namespaces:

namespaces.remove(black_namespace)

if len(namespaces)>0:

target_namespace = namespaces.pop(0)

else:

target_namespace = None

while target_namespace != None:

res = requests.get('https://{}/api/backlink/{}?namespace={}&from={}'.format(api_url,target_doc,target_namespace, backlink_value), headers = headers).json()

backlinks=res['backlinks']

backlink_value=res['from']

for doc in backlinks:

white=False

for white_doc in nontargets:

if doc['document'] == white_doc:

logger.info("화이트 리스트 문서 '{}' Pass".format(doc['document']))

white=True

break

if white:

continue

target_backlinks.append(doc['document'])

if backlink_value==None:

if len(namespaces)>0:

target_namespace = namespaces.pop(0)

backlink_value=''

else:

break

for doc in target_backlinks:

try:

res = requests.get('https://{}/api/edit/{}'.format(api_url,doc), headers = headers).json()

except:

res = requests.get('https://{}/api/edit/{}'.format(api_url,doc), headers = headers)

logger.log(logging.workingEmergencyStop, str(res.text))

streamlogger.log(logging.workingEmergencyStop, str(res.text))

exit()

try:

if not res['exists']:

continue

except:

logger.log(logging.editFail, ('문서명: {}'.format(doc),'res: {}'.format(res)))

streamlogger.log(logging.editFail, ('문서명: {}'.format(doc),'res: {}'.format(res)))

continue

token = res['token']

text = str(res['text'])

new_text = edit_link(doc, text)

res = requests.post('https://{}/api/edit/{}'.format(api_url,doc), json = {'text': new_text, 'log': log, 'token': token}, headers = headers)

try:

if res.json()['status'] != 'success':

logger.log(logging.editError, ('문서명: {}'.format(doc),'res: {}'.format(res.text)))

streamlogger.log(logging.editError, ('문서명: {}'.format(doc),'res: {}'.format(res.text)))

else:

data['edit'].append({

"doc": str(doc),

"rev": res.json()['rev'],

"text":text

})

logger.log(logging.editHistory,('문서명: {}'.format(doc),'rev: {}'.format(res.json()['rev'])))

except Exception as ex:

logger.critical(('문서명:',doc,'Error:',ex))

logger.log(logging.editError,(('문서명:',doc,'Error:',ex)))

streamlogger.critical(('문서명:',doc,'Error:',ex))

time.sleep(1)

with open('/content/drive/My Drive/Colab Notebooks/logs/{}/{}/edit_history.txt'.format(progStart_day,progStart_time), 'w') as outfile:

json.dump(data, outfile, indent=4, ensure_ascii=False)

def revert():

data={}

data['edit']=[]

json_data = {}

with open('/content/drive/My Drive/Colab Notebooks/logs/{}/{}/edit_history.txt'.format(revert_day,revert_time), "r") as json_file:

json_data = json.load(json_file)

for target in json_data['edit']:

doc = target['doc']

try:

res = requests.get('https://{}/api/edit/{}'.format(api_url,doc), headers = headers).json()

except:

res = requests.get('https://{}/api/edit/{}'.format(api_url,doc), headers = headers)

logger.log(logging.workingEmergencyStop, str(res.text))

streamlogger.log(logging.workingEmergencyStop, str(res.text))

exit()

try:

if not res['exists']:

continue

except:

logger.log(logging.editFail, ('문서명: {}'.format(doc),'res: {}'.format(res)))

streamlogger.log(logging.editFail, ('문서명: {}'.format(doc),'res: {}'.format(res)))

continue

token = res['token']

text = str(res['text'])

new_text = target['text']

log=log+'(r{}으로 되돌림)'.format(str(int(target['rev'])-1))

res = requests.post('https://{}/api/edit/{}'.format(api_url,doc), json = {'text': new_text, 'log': log, 'token': token}, headers = headers)

if res.json()['status'] != 'success':

logger.log(logging.editError, ('문서명: {}'.format(doc),'res: {}'.format(res.text)))

streamlogger.log(logging.editError, ('문서명: {}'.format(doc),'res: {}'.format(res.text)))

else:

data['edit'].append({

"doc": str(doc),

"rev": res.json()['rev'],

"text":text

})

logger.log(logging.editHistory,('문서명: {}'.format(doc),'rev: {}'.format(res.json()['rev'])))

time.sleep(1)

with open('/content/drive/My Drive/Colab Notebooks/logs/{}/{}/edit_history.txt'.format(progStart_day,progStart_time), 'w') as outfile:

json.dump(data, outfile, indent=4, ensure_ascii=False)

def customlinkchange():

data={}

data['edit']=[]

json_data = {}

for target, replace in targets:

for doc in custom_list:

try:

res = requests.get('https://{}/api/edit/{}'.format(api_url,doc), headers = headers).json()

except:

res = requests.get('https://{}/api/edit/{}'.format(api_url,doc), headers = headers)

logger.log(logging.workingEmergencyStop, str(res.text))

streamlogger.log(logging.workingEmergencyStop, str(res.text))

exit()

try:

if not res['exists']:

continue

except:

logger.log(logging.editFail, ('문서명: {}'.format(doc),'res: {}'.format(res)))

streamlogger.log(logging.editFail, ('문서명: {}'.format(doc),'res: {}'.format(res)))

continue

token = res['token']

text = str(res['text'])

new_text = edit_link(doc, text)

res = requests.post('https://{}/api/edit/{}'.format(api_url,doc), json = {'text': new_text, 'log': log, 'token': token}, headers = headers)

if res.json()['status'] != 'success':

logger.log(logging.editError, ('문서명: {}'.format(doc),'res: {}'.format(res.text)))

streamlogger.log(logging.editError, ('문서명: {}'.format(doc),'res: {}'.format(res.text)))

else:

data['edit'].append({

"doc": str(doc),

"rev": res.json()['rev'],

"text":text

})

logger.log(logging.editHistory,('문서명: {}'.format(doc),'rev: {}'.format(res.json()['rev'])))

time.sleep(1)

class runningThread(threading.Thread):

def __init__(self):

super().__init__()

self.state=False

def run(self):

self.state=True

logger.info('working start')

print('working start')

if mod=='normal':

backlinkchange()

elif mod=='revert':

if revert_day != '' or revert_day != '':

revert()

else:

print('revert 시간이 지정되지 않았습니다.')

log = '자동 편집 중 '

print('로그 변경됨:'+log+'(r{}으로 되돌림)'.format('int'))

elif mod=='custom':

customlinkchange()

else:

print('mod not select')

logger.info('working end')

print('working end')

self.state=False

####

checkDiscuss=checkWikiThread()

custom_thread=runningThread()

custom_thread.start()

while True:

try:

if not checkDiscuss.state:

checkDiscuss.run()

if not custom_thread.state or sys_stop:

logger.info('System Stop')

print('system stop')

break

if checkDiscuss.discuss_state != None:

logger.log(logging.workingEmergencyStop, 'discuss normal, System Emergency Stop')

streamlogger.log(logging.workingEmergencyStop, 'discuss normal, System Emergency Stop')

break

except Exception as ex:

logger.critical(ex)

streamlogger.critical(ex)

break

logging.shutdown()

reload(logging)