[주의!] 문서의 이전 버전(에 수정)을 보고 있습니다. 최신 버전으로 이동
이 사용자는 특수 권한을 가지고 있습니다.
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)):
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():
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:
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:
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))))
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
streamlogger.log(logging.discussError, ex)
html= await self.alterResponseTry(response,wait_time)
return html
def init(self):
self.state=False
self.discuss_state = None
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:
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)
#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
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:
#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
#content=link
#link=replace
#if anchor==None:
#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:
# 전용
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:
replace=target['replace']
if link==target_doc:
#내용이 없을경우 원래 링크가 내용이 됨
#content_replace bool 추가
if content==None and content_replace:
#content_replace bool 추가
if content==None and content_replace:
#혹시 모를 사고 예방
if detachBar==None:
if detachBar==None:
content='|{}'.format(link)
else:
content='{}'.format(link)
elif content==None:
content=''
link=replace
return '{}{}{}{}{}'.format(first,link,middle,content,last)
return '{}{}{}{}{}'.format(first,link,middle,content,last)
return match_obj.group(0)
def checkchange(match_obj):
def checkdelete(match_obj):
#분류 정리 전용
link = match_obj.group('link')
anchor = match_obj.group('anchor')
content = match_obj.group('content')
for target in targets:
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:
replace=target['replace']
if link==target_doc:
if content==None:
#내용이 없을경우 원래 링크가 내용이 됨
content=link
content=link
link=replace
if anchor==None:
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>rdd=)(?P<link>((\,)|(\\)|(\))|[^,)])*)(?P<last>[,)])")
detailContent_patten=re.compile(r"(?P<first>문서명=)(?P<link>((\,)|(\\)|(\))|[^,)])*)(?P<last>[,)])")
categoryMean_patten=re.compile(r"(?P<first>nd=)(?P<link>((\,)|(\\)|(\))|[^,)])*)(?P<last>[,)])")
top_patten=re.compile(r"(?P<first>topd=)(?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)")
otherMean_patten=re.compile(r"(?P<first>rdd=)(?P<link>((\,)|(\\)|(\))|[^,)])*)(?P<last>[,)])")
detailContent_patten=re.compile(r"(?P<first>문서명=)(?P<link>((\,)|(\\)|(\))|[^,)])*)(?P<last>[,)])")
categoryMean_patten=re.compile(r"(?P<first>nd=)(?P<link>((\,)|(\\)|(\))|[^,)])*)(?P<last>[,)])")
top_patten=re.compile(r"(?P<first>topd=)(?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:
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:
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']:
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:
backlinks=res['backlinks']
backlink_value=res['from']
for doc in backlinks:
white=False
for white_doc in nontargets:
for white_doc in nontargets:
if doc['document'] == white_doc:
logger.info("화이트 리스트 문서 '{}' Pass".format(doc['document']))
white=True
break
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=''
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()
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
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)))
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'])))
"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))
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:
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:
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']:
for target in json_data['edit']:
doc = target['doc']
try:
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()
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
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)))
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'])))
"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:
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()
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
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)))
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'])))
"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
self.state=False
def run(self):
self.state=True
logger.info('working start')
print('working start')
if mod=='normal':
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'))
print('로그 변경됨:'+log+'(r{}으로 되돌림)'.format('int'))
elif mod=='custom':
customlinkchange()
else:
print('mod not select')
logger.info('working end')
print('working end')
self.state=False
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
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
streamlogger.log(logging.workingEmergencyStop, 'discuss normal, System Emergency Stop')
break
except Exception as ex:
logger.critical(ex)
streamlogger.critical(ex)
break
streamlogger.critical(ex)
break
logging.shutdown()
reload(logging)
reload(logging)