Python」タグアーカイブ

GExiv2 and subprocess

前回の GExiv2 の件。
comipoli に実装しようとしてハマったので覚書。

GSubprocess から GUnixInputStream を得て GExiv2 に読ませる。
その GUnixInputStream からは GdkPixbuf を作ることができない。
順番を逆にしても駄目、Stream の再利用はできないみたい。

他の手段で GExiv2 を使おうと思ったけど上手くいかない。
他の手段で GdkPixbuf を得ることは無理。
exif 取得と画像取得で別にアクセスするしかないようだ。
遅くなるだろうけど気にならないレベルならいいかなって。

しかし GSubprocess をもう一つ使をうとするもエラー。
原因は解らない、GSubprocess の情報なんて皆無だ。

失敗をズラズラ書いても無意味なので結論。
GSubprocess と subprocess を両方使う強行手段でなんとかなった。

class ComipoliArchive:
    def __init__(self):
    	# etc...

    def _zip_escape(self, filename):
        ESCAPE = '[]*?!^-\\'
        res = ''
        for s in filename:
            if s in ESCAPE:
                res += '\\'
            res += s
        return res

    def _on_wait(self, proc, res):
        proc.wait_check_finish(res)

    def __getitem__(self, num):
        if num == self.max:
            raise IndexError
        try:
            args = ['unzip', '-pj', self.path, self._zip_escape(self.namelist[num])]
            # tag
            ori = 0
            pr = subprocess.Popen(args, encoding='UTF-8', stdout=subprocess.PIPE, stderr=subprocess.PIPE)
            fd = pr.stdout.fileno()
            with open(fd, 'rb') as f:
                metadata = GExiv2.Metadata()
                metadata.open_buf(f.read())
                ori = metadata.get_orientation()
            pr.wait()
            # pixbuf
            sp = Gio.Subprocess.new(args, Gio.SubprocessFlags.STDOUT_PIPE | Gio.SubprocessFlags.STDERR_MERGE)
            sp.wait_check_async(None, self._on_wait)
            stream = sp.get_stdout_pipe()
            p = GdkPixbuf.Pixbuf.new_from_stream(stream)
            if ori == GExiv2.Orientation.HFLIP:
                p = p.flip(True)
            elif ori == GExiv2.Orientation.ROT_180:
                p = p.rotate_simple(GdkPixbuf.PixbufRotation.UPSIDEDOWN)
            elif ori == GExiv2.Orientation.VFLIP:
                p = p.flip(False)
            elif ori == GExiv2.Orientation.ROT_90_HFLIP:
                p = p.rotate_simple(GdkPixbuf.PixbufRotation.CLOCKWISE).flip(True)
            elif ori == GExiv2.Orientation.ROT_90:
                p = p.rotate_simple(GdkPixbuf.PixbufRotation.CLOCKWISE)
            elif ori == GExiv2.Orientation.ROT_90_VFLIP:
                p = p.rotate_simple(GdkPixbuf.PixbufRotation.CLOCKWISE).flip(False)
            elif ori == GExiv2.Orientation.ROT_270:
                p = p.rotate_simple(GdkPixbuf.PixbufRotation.COUNTERCLOCKWISE)
            stream.close()
            sp.force_exit()
            return p
        except Exception as e:
            raise e

前回の cbz を開いてみる。

orientation

よし回転されているな。
注意点は ROT_90 は 90 度傾いているから 270 度回転させるということ。
戻す度数という意味ではない、flip は手持ちサンプルが無いのでこれで正しいか未確認。
後 Popen は wait を忘れずに。

Scaling: GDK-PixBuf Reference Manual

展開速度も別に気にならない、てか何か変わったか?のレベル。
多分 GExiv2 が凄いだけなんだろうけど。

1200px

PC での twitter の画像観覧が少し前から原寸になった。
twitter の 4 コマ漫画が大好きな筆者は少し困ったことに。

少し前まで普通にコンテキストメニューから落とすと縦 1200px 固定だった。
原寸を落とす方法も知っていたけど cbz にする場合このサイズが都合よかった。

つまりその時作った cbz が沢山ある。
それも完結していない続き物が多い。

続き物のページを cbz に追加する場合は縦 1200px に縮小しないと整合性が。
原寸は巨大なものが多いので表示に時間が掛かるというのもある。
途中のページから急に重く、なんて嫌だよ。

そういえば追加する時に以前追加した最後の名前は何だったか調べるのも面倒。
020.jpg だったら 021.jpg にして追加したい、ずっとそうしているし。

画像を縮小して勝手に cbz ファイルから名前を調べて追加。
なんて定型作業を自動化したいな。

だったら自分で作ればいいじゃないか!

#!/usr/bin/env python3

'''
    ダウンロードした続き物の twitter 4コマまんがを選択して使う
    ダイアログで選択した cbz ファイルの 001.jpg から始まる名前を取得
    名前順の最後になるよう 022.jpg 等の名前を付ける
    縦 1200px に画像をサイズダウンしキャッシュに保存
    それを cbz ファイルに追加
    までを自動化する Nautilus スクリプト
'''

import os, re, zipfile, gi
gi.require_version('Gtk', '3.0')
gi.require_version('GdkPixbuf', '2.0')
from gi.repository import GLib, Gtk, GdkPixbuf

# move ~/.catch
cachedir = GLib.get_user_cache_dir()
os.chdir(cachedir)

dlg = Gtk.FileChooserNative(title='Open', action=Gtk.FileChooserAction.OPEN)
dlg.set_current_folder(GLib.get_home_dir())
ft = Gtk.FileFilter()
ft.set_name('Comic Book Archive')
ft.add_mime_type('application/x-cbz')
ft.add_mime_type('application/vnd.comicbook+zip')
dlg.add_filter(ft)
r = dlg.run()
dlg.destroy()
if r == Gtk.ResponseType.ACCEPT:
    arc_name = dlg.get_file().get_path()
    with zipfile.ZipFile(arc_name, 'a') as z:
        # last name
        s = z.namelist()[-1]
        num = int(os.path.splitext(s)[0])
        # loop
        path_array = os.environ['NAUTILUS_SCRIPT_SELECTED_FILE_PATHS'].split('\n')
        for filepath in path_array:
            try:
                pixbuf = GdkPixbuf.Pixbuf.new_from_file(filepath)
            except:
                continue
            # resize
            p_width = pixbuf.get_width()
            p_height = pixbuf.get_height()
            width = p_width * 1200 // p_height
            smallpix = pixbuf.scale_simple(width, 1200, GdkPixbuf.InterpType.BILINEAR)
            # create name
            ext = os.path.splitext(filepath)[1]
            num += 1
            name = f'{num:03d}{ext}'
            # create jpeg or png
            if re.search(r'\.(jpg|jpeg)$', filepath, re.I):
                smallpix.savev(name, 'jpeg', ['quality'], ['85'])
            elif re.search(r'\.png$', filepath, re.I):
                smallpix.savev(name, 'png', ['compression'], ['9'])
            z.write(name)

作ってみた。

file-roller で一旦 cbz 内のファイル名を調べる必要が無くなった。
1.jpg とか適当な名前の一時保存でも追加時に自動変名、我ながら超便利。
何故今までコレを思いつかなかったのだ俺よ。

ZipFile 作成にうっかり w 指定をして消えてしまったファイルも…
こういうのを作る時はバックアップをしてから、絶対だよ。

しかし特に Node.js 屋で見かけるんだけーが。
「コードを書くのが面倒だから npm でインストール!」
みたいなことを書く輩はいったい何故プログラミングを勉強しているのだ?
定型作業が面倒だからコードを書いて楽をするんじゃないのかと。
コードを書くのが面倒な人がそんなことするとは思えない。
定型作業はタイピング速度で解決、とか考えそう。

os.forkpty

あけましておめでとうございます。
元旦といえば、そうプログラミングです。

ということで、Python で sshpass を再現の話。
os.forkpty というものがあるようだ。

python – Using os.forkpty() to create a pseudo-terminal to ssh to a remote server and communicate with it – Stack Overflow

これだよコレ、似たようなことを考える人っているもんだ。
がんばって sshpass と完全に同じコードにしなくてもよかった。
というか ptmx では上手くいかない、擬似端末ならこっちでもいいはず。

Python3 に書き換えしなきゃね。
色々問題はあるけどなんとかなったコードをとりあえず。

#!/usr/bin/env python3

import os, sys, signal, time

# var
HOSTNAME = 'username@hostname'
PASSWORD = '********'

# Ctrl+C
signal.signal(signal.SIGINT, signal.SIG_DFL)

pid, fd = os.forkpty()

def cmd(s):
    '''
        長い出力の場合分割される場合がある
        os.read はこの場合ループにすると値を戻さずフリーズする
        しかたがないので空打ちの場合は残りの読み込みにしている
    '''
    if s:
        os.write(fd, f'{s}\n'.encode('utf-8'))
        time.sleep(1)
        res = os.read(fd, 1024).decode()
        # 一行目を取り除いて表示
        sys.stdout.write(res[res.find('\n')+1 : ])
    else:
        res = os.read(fd, 1024).decode()
        sys.stdout.write(res)
    sys.stdout.flush()

if pid == 0:
    #os.setsid() # Error
    os.execvp('ssh',['ssh', HOSTNAME])
    # 親プロセスに切り替わるので以下は実行されない
    print('execvp Error!')

# レスポンスが遅いと表示されないけどログインは可能
output = os.read(fd, 1024)
sys.stdout.write(output.decode())
sys.stdout.flush()
# パスいワード
os.write(fd, f'{PASSWORD}\n'.encode('utf-8'))
# 只の時間稼ぎ
time.sleep(1)
res = os.read(fd, 1024).decode()
# 最初のプロンプトが表示できないので仮プロンプト
print(f'{res}First Command > ', end='')
# exit で終了
while True:
    s = input()
    if s == 'exit':
        break
    cmd(s)
os.write(fd, 'exit\n'.encode('utf-8'))
print('__DONE__')

よしこれで sshpass 不要でパスワード入力ができる。
しかし time.sleep を駆使するしかないのかな?
今のところこんな感じ。

調子こいて pysshpass みたいなものを作ろうと思ったけど…
os.forkpty では setsid できないし ptmx だと暴走するしetc…
我が Macbook Air の CPU ファン全開音なんて初めて聞いたよ。
ま、必要ないか。

しかし思っていたより Fedora と macOS では使える関数が違う。
os.fsync は macOS で問題ないけど Fedora は何をやってもエラー。
fcntl でノンブロッキングI/O は macOS ではエラー出まくり。
他の Linux や BSD では、なんて考えたくないな。

ptmx

sftp アプリを PyObjC で作るって話。
やってます、上手くいかないだけです。
とりあえず今やっていること。

sshpass/main.c at master ? bauruine/sshpass ? GitHub

sshpass のソースってたったコレだけなんだよね。
これなら Python3 の標準モジュールだけで模写できそうだと考えた。
UNIX の中身の勉強にもなるし一石二鳥。

/dev/ptmx って何だ?

Ubuntu Manpage: ptmx, pts – 擬似端末のマスタとスレーブ

たしかに sudo や ssh は stdin やパイプからの入力はできない。
/dev/ptmx を利用すればなんとかなるってことね。

#!/usr/bin/env python3

import os, signal, fcntl

# Ctrl+C
signal.signal(signal.SIGINT, signal.SIG_DFL)

# ptmx
masterpt = os.open('/dev/ptmx', os.O_RDWR)

# not macOS
#fcntl.fcntl(masterpt, fcntl.F_SETFL, os.O_NONBLOCK)
#name = os.ttyname(masterpt)

childpid = os.fork()
if childpid == 0:
    os.setsid()
    #os.execvp('ssh', ['ssh', 'sasakima-nao@mba2.local'])
    os.execvp('python3', ['python3'])
    # Error
    print(f'{__file__}: Failed to run command')

#slavept = os.open(name, os.O_RDWR|os.O_NOCTTY)

os.close(masterpt)
#os.close(slavept)
print(f'{__file__}: EOL')

とりあえず fork して execvp することはできた。
呼び出し側のプロセスを終了して子プロセスに置き換える。
ほぼ os モジュールだけでなんとかなるのようで。

Man page of EXEC

起動すると EOL が表示され Python3 インタプリタが始まる。
Ctrl+D すると bash はもう破棄されているので gnome-terminal が終了する。
macOS では「プロセスが完了しました」と出る。

で、ssh のほうの現状。

こんなになってしまう。
sshpass.c をトレースしているつもりだけどまだ何か間違っているっぽい。
上手くいったらまた今度。

subprocess run Popen

そういえば最近このブログは subprocess.Popen を使っているけど。
subprocess.run で全部まかなえるようになったんじゃなかったっけ?
Popen を使う理由は何だろう。

subprocessでPythonからLinuxコマンド実行

あぁ戻り値を得ないならば非同期実行になるってことね。
しかしなんともなサンプルコードだ、多分 Windows なんだろうけど。
UNIX 系ならこんなに単純なサンプルコードにできる。

#!/usr/bin/env python3

import subprocess

subprocess.Popen(['sh', '-c', 'sleep 3; echo First'])
subprocess.Popen(['sh', '-c', 'echo Second'])

print('__done__')

__done__ が一番最初に表示されるね、なるほど。
ただし with as を使うと戻り値を使うのと同様になるので順番どおりになる。

#!/usr/bin/env python3

import subprocess

with subprocess.Popen(['sh', '-c', 'sleep 3; echo First']) as p:
    pass
with subprocess.Popen(['sh', '-c', 'echo Second']) as q:
    pass

print('__done__')

この特性を上手く利用すれば色々便利になりそう。