"""실행 중인 밀리 PDF 연결 GUI. 앱 실행/종료 및 PDF 파일 변경 없음."""

import json

from pathlib import Path

import queue

import re

import subprocess

import threading

import time

import tkinter as tk

from tkinter import ttk

from urllib.parse import urlparse, parse_qs, unquote

import urllib.request

 

 

def load_hook():

"""훅 스크립트를 읽어 온다. 실패는 호출한 쪽에서 처리하게 예외를 그대로 올린다."""

return Path(__file__).with_name('millie_gui_hook.js').read_text(encoding='utf-8')

 

 

def find_readers():

command = (

"[Console]::OutputEncoding=[System.Text.Encoding]::UTF8; "

"Get-CimInstance Win32_Process | Where-Object { "

"$_.Name -eq 'msedgewebview2.exe' -and "

"$_.CommandLine -like '*millie-desktop*' } | "

"Select-Object CommandLine | ConvertTo-Json -Compress"

)

result = subprocess.run(

['powershell.exe', '-NoProfile', '-Command', command],

capture_output=True, encoding='utf-8-sig', errors='replace',

creationflags=subprocess.CREATE_NO_WINDOW, timeout=20,

)

if result.returncode:

raise RuntimeError('밀리 프로세스를 확인하지 못했습니다: ' + result.stderr.strip())

rows = json.loads(result.stdout) if result.stdout.strip() else []

if isinstance(rows, dict):

rows = [rows]

ports = set()

for row in rows:

match = re.search(r'--remote-debugging-port=(\d+)', row.get('CommandLine') or '')

if match and int(match[1]):

ports.add(int(match[1]))

if not ports:

raise RuntimeError(

'밀리의 디버깅 연결을 찾지 못했습니다.\n'

'디버깅이 활성화된 밀리에서 PDF 책을 연 뒤 다시 확인하세요.\n'

'이 프로그램은 밀리를 재실행하지 않습니다.'

)

opener = urllib.request.build_opener(urllib.request.ProxyHandler({}))

readers, errors = [], []

for port in ports:

try:

with opener.open(f'http://127.0.0.1:{port}/json/list', timeout=3) as response:

targets = json.load(response)

for target in targets:

url = urlparse(target.get('url', ''))

if (target.get('type') == 'page' and url.hostname in ('localhost', '127.0.0.1')

and parse_qs(url.query).get('bookSeq') and target.get('webSocketDebuggerUrl')):

readers.append(target)

except (OSError, ValueError) as error:

errors.append(str(error))

if not readers:

raise RuntimeError('열린 PDF 뷰어를 찾지 못했습니다. 밀리에서 PDF 책을 열어 주세요.'

+ ('\n연결 오류: ' + '; '.join(errors) if errors else ''))

return readers

 

 

class CDP:

def __init__(self, url):

try:

import websocket

except ImportError as error:

raise RuntimeError('필요한 모듈을 설치하세요: python -m pip install websocket-client') from error

# recv 타임아웃 예외를 나중에 구분해서 삼키기 위해 클래스를 붙잡아 둔다.

self._timeout_exc = websocket.WebSocketTimeoutException

self.ws = websocket.create_connection(

url, timeout=5, suppress_origin=True, http_no_proxy=['localhost', '127.0.0.1'])

self.seq = 0

 

def call(self, method, params=None):

self.seq += 1

self.ws.send(json.dumps({'id': self.seq, 'method': method, 'params': params or {}}))

deadline = time.monotonic() + 8

# recv 자체 타임아웃(짧게)을 반복하며 deadline까지 기다린다.

# 이렇게 해야 소켓 타임아웃이 아니라 의도한 8초 TimeoutError로 끝난다.

while True:

remaining = deadline - time.monotonic()

if remaining <= 0:

raise TimeoutError('뷰어 응답 시간이 초과됐습니다.')

self.ws.settimeout(min(1.0, max(0.1, remaining)))

try:

reply = json.loads(self.ws.recv())

except self._timeout_exc:

continue # 아직 응답 전 — deadline 안이면 계속 기다린다.

if reply.get('id') == self.seq:

if 'error' in reply:

raise RuntimeError(reply['error'].get('message', str(reply['error'])))

return reply.get('result', {})

 

def evaluate(self, expression):

reply = self.call('Runtime.evaluate', {'expression': expression, 'returnByValue': True})

if reply.get('exceptionDetails'):

raise RuntimeError('뷰어 상태를 읽지 못했습니다.')

return reply.get('result', {}).get('value')

 

def close(self):

self.ws.close()

 

 

def capture(target, hook, status=lambda message: None, cancelled=None):

cancelled = cancelled or threading.Event()

client = CDP(target['webSocketDebuggerUrl'])

script_id = None

try:

state = client.evaluate('({url:location.href, page:window.pdf?.currentPage, '

'supported:typeof window.PSPDF === "function" || '

'typeof window.PDF === "function"})')

if state['url'] != target['url']:

raise RuntimeError('책이 변경됐습니다. 열린 책 목록을 다시 확인하세요.')

if not state.get('supported'):

raise RuntimeError('지원되는 PDF 뷰어가 아닙니다. PDF 책의 로딩이 끝난 뒤 다시 시도하세요.')

cached = client.evaluate('window.__millieGuiResult || null')

if cached and cached.get('hasPassword'):

return cached

client.call('Page.enable')

script_id = client.call('Page.addScriptToEvaluateOnNewDocument', {'source': hook})['identifier']

status('현재 PDF 화면을 새로고침하고 있습니다…')

client.call('Page.reload')

deadline = time.monotonic() + 35

payload = None

restored = False

while time.monotonic() < deadline:

if cancelled.is_set():

raise RuntimeError('캡처를 중단했습니다.')

data = client.evaluate('({capture:window.__millieGuiResult || null, '

'ready:!!window.pdf?.instance, url:location.href})')

if not data:

time.sleep(0.2)

continue

if data['url'] != state['url']:

raise RuntimeError('캡처 중 다른 책으로 이동했습니다. 다시 시도하세요.')

payload = data.get('capture') or payload

if data.get('ready'):

page = state.get('page')

if isinstance(page, int) and not restored:

client.evaluate(f'window.pdf.setPage({page})')

restored = True

if payload:

if not payload.get('hasPassword'):

raise RuntimeError('이 PDF의 초기화 과정에서는 암호가 전달되지 않았습니다.')

return payload

time.sleep(0.25)

if payload and payload.get('hasPassword'):

payload['warning'] = '암호는 캡처했지만 읽던 위치의 복원은 확인하지 못했습니다.'

return payload

raise RuntimeError('암호 전달을 감지하지 못했습니다. PDF가 정상적으로 열리는지 확인하세요.')

finally:

if script_id:

try:

client.call('Page.removeScriptToEvaluateOnNewDocument', {'identifier': script_id})

except Exception:

pass

client.close()

 

 

class App:

def __init__(self, root, hook, hook_error=None):

self.root = root

self.hook = hook

self.hook_error = hook_error

self.events = queue.Queue()

self.cancelled = threading.Event()

self.targets = []

self.busy = False

root.title('밀리 PDF 암호 확인')

root.geometry('760x420')

root.minsize(650, 390)

root.option_add('*Font', ('맑은 고딕', 10))

frame = ttk.Frame(root, padding=22)

frame.pack(fill='both', expand=True)

ttk.Label(frame, text='현재 열린 PDF 암호', font=('맑은 고딕', 17, 'bold')).pack(anchor='w')

ttk.Label(frame, text='밀리에서 PDF 책을 연 뒤 선택하세요. 앱은 재실행하지 않습니다.').pack(anchor='w', pady=(8, 4))

ttk.Label(frame, text='첫 캡처 시 PDF 화면만 새로고침하며 읽던 위치를 복원합니다.').pack(anchor='w')

row = ttk.Frame(frame)

row.pack(fill='x', pady=(18, 12))

self.books = ttk.Combobox(row, state='readonly')

self.books.pack(side='left', fill='x', expand=True)

self.books.bind('<<ComboboxSelected>>', lambda event: self.clear_result())

self.refresh_button = ttk.Button(row, text='열린 책 확인', command=self.refresh)

self.refresh_button.pack(side='left', padx=(8, 0))

self.capture_button = ttk.Button(frame, text='선택한 PDF 암호 확인', command=self.start_capture)

self.capture_button.pack(anchor='w')

self.filename = tk.StringVar(value='파일: —')

ttk.Label(frame, textvariable=self.filename).pack(anchor='w', pady=(16, 5))

self.password = tk.StringVar()

ttk.Entry(frame, textvariable=self.password, state='readonly', font=('Consolas', 10)).pack(fill='x')

self.copy_button = ttk.Button(frame, text='암호 복사', command=self.copy, state='disabled')

self.copy_button.pack(anchor='e', pady=8)

self.status = tk.StringVar(value='열린 책을 확인하고 있습니다…')

ttk.Label(frame, textvariable=self.status, wraplength=690, justify='left').pack(anchor='w')

root.protocol('WM_DELETE_WINDOW', self.close)

root.after(100, self.poll)

# 훅 로딩이 실패했으면 앱은 뜨되 이유를 알리고 캡처는 막는다.

if self.hook is None:

self.status.set('훅 스크립트를 불러오지 못해 암호 확인을 사용할 수 없습니다.\n'

+ (self.hook_error or '')

+ '\n(millie_gui_hook.js 파일이 이 프로그램과 같은 폴더에 있어야 합니다.)')

self.capture_button.configure(state='disabled')

else:

root.after(200, self.refresh)

 

def clear_result(self):

self.password.set('')

self.filename.set('파일: —')

self.copy_button.configure(state='disabled')

 

def work(self, operation, kind):

if self.busy:

return

self.busy = True

self.refresh_button.configure(state='disabled')

self.capture_button.configure(state='disabled')

self.books.configure(state='disabled')

def run():

try:

self.events.put((kind, operation()))

except Exception as error:

self.events.put(('error', str(error)))

finally:

self.events.put(('done', None))

threading.Thread(target=run, daemon=True).start()

 

def refresh(self):

if self.busy or self.hook is None:

return

self.clear_result()

self.targets = []

self.books.set('')

self.books.configure(values=[])

self.status.set('열린 책을 확인하고 있습니다…')

self.work(find_readers, 'readers')

 

def start_capture(self):

if self.hook is None:

self.status.set('훅 스크립트를 불러오지 못해 암호 확인을 사용할 수 없습니다.')

return

index = self.books.current()

if self.busy or not 0 <= index < len(self.targets):

self.status.set('먼저 열린 PDF 책을 선택하세요.')

return

self.clear_result()

self.status.set('뷰어에 연결하고 있습니다…')

hook = self.hook

self.work(lambda: capture(self.targets[index], hook,

lambda message: self.events.put(('status', message)), self.cancelled), 'capture')

 

def poll(self):

# 이벤트 처리 중 어떤 예외가 나도 폴링 재예약(finally)은 반드시 실행되게 한다.

try:

while not self.events.empty():

kind, value = self.events.get_nowait()

if kind == 'done':

self.busy = False

self.refresh_button.configure(state='normal')

self.capture_button.configure(

state='normal' if (self.targets and self.hook) else 'disabled')

self.books.configure(state='readonly')

elif kind in ('error', 'status'):

self.status.set(value)

elif kind == 'readers':

self.targets = value

self.books.configure(values=[t.get('title') or t['url'] for t in value])

self.books.current(0)

self.status.set('암호를 확인할 PDF를 선택하고 버튼을 누르세요.')

elif kind == 'capture':

self.password.set(value['password'])

self.filename.set('파일: ' + unquote(urlparse(value.get('url', '')).path.rsplit('/', 1)[-1]))

self.copy_button.configure(state='normal')

self.status.set(value.get('warning') or '암호를 캡처했습니다. 읽던 PDF 화면을 유지합니다.')

except Exception as error:

# 개별 이벤트가 깨져도 앱 전체가 멈추지 않게 상태창에만 알린다.

try:

self.status.set('화면 갱신 중 오류가 발생했습니다: ' + str(error))

except Exception:

pass

finally:

self.root.after(100, self.poll)

 

def copy(self):

if self.password.get():

self.root.clipboard_clear()

self.root.clipboard_append(self.password.get())

self.status.set('암호를 복사했습니다.')

 

def close(self):

self.cancelled.set()

self.root.destroy()

 

 

if __name__ == '__main__':

root = tk.Tk()

try:

hook = load_hook()

hook_error = None

except OSError as error:

hook = None

hook_error = str(error)

App(root, hook, hook_error)

root.mainloop()