Upstream
/
gibMacOS
mirror of https://github.com/corpnewt/gibMacOS.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Add files via upload

This commit is contained in:
CorpNewt 2018-10-06 22:07:05 -05:00 committed by GitHub
parent 79a23ca60a
commit b2896f5304
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
7 changed files with 1217 additions and 535 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
Scripts/__init__.py
View File

@ -1,4 +1,4 @@
from os.path import dirname, basename, isfile
import glob
modules = glob.glob(dirname(__file__)+"/*.py")
from os.path import dirname, basename, isfile
import glob
modules = glob.glob(dirname(__file__)+"/*.py")
__all__ = [ basename(f)[:-3] for f in modules if isfile(f) and not f.endswith('__init__.py')]

436
Scripts/disk.py Normal file
View File

@ -0,0 +1,436 @@
import subprocess, plistlib, sys, os, time, json
sys.path.append(os.path.abspath(os.path.dirname(os.path.realpath(__file__))))
import run
class Disk:
def __init__(self):
self.r = run.Run()
self.diskutil = self.get_diskutil()
self.os_version = ".".join(
self.r.run({"args":["sw_vers", "-productVersion"]})[0].split(".")[:2]
)
self.apfs = {}
self._update_disks()
def _get_plist(self, s):
p = {}
try:
if sys.version_info >= (3, 0):
p = plistlib.loads(s.encode("utf-8"))
else:
p = plistlib.readPlistFromString(s.encode("utf-8"))
except:
pass
return p
def _compare_versions(self, vers1, vers2):
# Helper method to compare ##.## strings
#
# vers1 < vers2 = True
# vers1 = vers2 = None
# vers1 > vers2 = False
#
try:
v1_parts = vers1.split(".")
v2_parts = vers2.split(".")
except:
# Formatted wrong - return None
return None
for i in range(len(v1_parts)):
if int(v1_parts[i]) < int(v2_parts[i]):
return True
elif int(v1_parts[i]) > int(v2_parts[i]):
return False
# Never differed - return None, must be equal
return None
def update(self):
self._update_disks()
def _update_disks(self):
self.disks = self.get_disks()
self.disk_text = self.get_disk_text()
if self._compare_versions("10.12", self.os_version):
self.apfs = self.get_apfs()
else:
self.apfs = {}
def get_diskutil(self):
# Returns the path to the diskutil binary
return self.r.run({"args":["which", "diskutil"]})[0].split("\n")[0].split("\r")[0]
def get_disks(self):
# Returns a dictionary object of connected disks
disk_list = self.r.run({"args":[self.diskutil, "list", "-plist"]})[0]
return self._get_plist(disk_list)
def get_disk_text(self):
# Returns plain text listing connected disks
return self.r.run({"args":[self.diskutil, "list"]})[0]
def get_disk_info(self, disk):
disk_id = self.get_identifier(disk)
if not disk_id:
return None
disk_list = self.r.run({"args":[self.diskutil, "info", "-plist", disk_id]})[0]
return self._get_plist(disk_list)
def get_disk_fs(self, disk):
disk_id = self.get_identifier(disk)
if not disk_id:
return None
return self.get_disk_info(disk_id).get("FilesystemName", None)
def get_disk_fs_type(self, disk):
disk_id = self.get_identifier(disk)
if not disk_id:
return None
return self.get_disk_info(disk_id).get("FilesystemType", None)
def get_apfs(self):
# Returns a dictionary object of apfs disks
output = self.r.run({"args":"echo y | " + self.diskutil + " apfs list -plist", "shell" : True})
if not output[2] == 0:
# Error getting apfs info - return an empty dict
return {}
disk_list = output[0]
p_list = disk_list.split("<?xml")
if len(p_list) > 1:
# We had text before the start - get only the plist info
disk_list = "<?xml" + p_list[-1]
return self._get_plist(disk_list)
def is_apfs(self, disk):
disk_id = self.get_identifier(disk)
if not disk_id:
return None
# Takes a disk identifier, and returns whether or not it's apfs
for d in self.disks.get("AllDisksAndPartitions", []):
if not "APFSVolumes" in d:
continue
if d.get("DeviceIdentifier", "").lower() == disk_id.lower():
return True
for a in d.get("APFSVolumes", []):
if a.get("DeviceIdentifier", "").lower() == disk_id.lower():
return True
return False
def is_apfs_container(self, disk):
disk_id = self.get_identifier(disk)
if not disk_id:
return None
# Takes a disk identifier, and returns whether or not that specific
# disk/volume is an APFS Container
for d in self.disks.get("AllDisksAndPartitions", []):
# Only check partitions
for p in d.get("Partitions", []):
if disk_id.lower() == p.get("DeviceIdentifier", "").lower():
return p.get("Content", "").lower() == "apple_apfs"
return False
def is_cs_container(self, disk):
disk_id = self.get_identifier(disk)
if not disk_id:
return None
# Takes a disk identifier, and returns whether or not that specific
# disk/volume is an CoreStorage Container
for d in self.disks.get("AllDisksAndPartitions", []):
# Only check partitions
for p in d.get("Partitions", []):
if disk_id.lower() == p.get("DeviceIdentifier", "").lower():
return p.get("Content", "").lower() == "apple_corestorage"
return False
def is_core_storage(self, disk):
disk_id = self.get_identifier(disk)
if not disk_id:
return None
if self._get_physical_disk(disk_id, "Logical Volume on "):
return True
return False
def get_identifier(self, disk):
# Should be able to take a mount point, disk name, or disk identifier,
# and return the disk's identifier
# Iterate!!
if not disk or not len(str(disk)):
return None
disk = disk.lower()
if disk.startswith("/dev/r"):
disk = disk[len("/dev/r"):]
elif disk.startswith("/dev/"):
disk = disk[len("/dev/"):]
if disk in self.disks.get("AllDisks", []):
return disk
for d in self.disks.get("AllDisksAndPartitions", []):
for a in d.get("APFSVolumes", []):
if disk in [ a.get(x, "").lower() for x in ["DeviceIdentifier", "VolumeName", "VolumeUUID", "DiskUUID", "MountPoint"] ]:
return a.get("DeviceIdentifier", None)
for a in d.get("Partitions", []):
if disk in [ a.get(x, "").lower() for x in ["DeviceIdentifier", "VolumeName", "VolumeUUID", "DiskUUID", "MountPoint"] ]:
return a.get("DeviceIdentifier", None)
# At this point, we didn't find it
return None
def get_top_identifier(self, disk):
disk_id = self.get_identifier(disk)
if not disk_id:
return None
return disk_id.replace("disk", "didk").split("s")[0].replace("didk", "disk")
def _get_physical_disk(self, disk, search_term):
# Change disk0s1 to disk0
our_disk = self.get_top_identifier(disk)
our_term = "/dev/" + our_disk
found_disk = False
our_text = ""
for line in self.disk_text.split("\n"):
if line.lower().startswith(our_term):
found_disk = True
continue
if not found_disk:
continue
if line.lower().startswith("/dev/disk"):
# At the next disk - bail
break
if search_term.lower() in line.lower():
our_text = line
break
if not len(our_text):
# Nothing found
return None
our_stores = "".join(our_text.strip().split(search_term)[1:]).split(" ,")
if not len(our_stores):
return None
for store in our_stores:
efi = self.get_efi(store)
if efi:
return store
return None
def get_physical_store(self, disk):
# Returns the physical store containing the EFI
disk_id = self.get_identifier(disk)
if not disk_id:
return None
if not self.is_apfs(disk_id):
return None
return self._get_physical_disk(disk_id, "Physical Store ")
def get_core_storage_pv(self, disk):
# Returns the core storage physical volume containing the EFI
disk_id = self.get_identifier(disk)
if not disk_id:
return None
if not self.is_core_storage(disk_id):
return None
return self._get_physical_disk(disk_id, "Logical Volume on ")
def get_parent(self, disk):
# Disk can be a mount point, disk name, or disk identifier
disk_id = self.get_identifier(disk)
if self.is_apfs(disk_id):
disk_id = self.get_physical_store(disk_id)
elif self.is_core_storage(disk_id):
disk_id = self.get_core_storage_pv(disk_id)
if not disk_id:
return None
if self.is_apfs(disk_id):
# We have apfs - let's get the container ref
for a in self.apfs.get("Containers", []):
# Check if it's the whole container
if a.get("ContainerReference", "").lower() == disk_id.lower():
return a["ContainerReference"]
# Check through each volume and return the parent's container ref
for v in a.get("Volumes", []):
if v.get("DeviceIdentifier", "").lower() == disk_id.lower():
return a.get("ContainerReference", None)
else:
# Not apfs - go through all volumes and whole disks
for d in self.disks.get("AllDisksAndPartitions", []):
if d.get("DeviceIdentifier", "").lower() == disk_id.lower():
return d["DeviceIdentifier"]
for p in d.get("Partitions", []):
if p.get("DeviceIdentifier", "").lower() == disk_id.lower():
return d["DeviceIdentifier"]
# Didn't find anything
return None
def get_efi(self, disk):
disk_id = self.get_parent(self.get_identifier(disk))
if not disk_id:
return None
# At this point - we should have the parent
for d in self.disks["AllDisksAndPartitions"]:
if d.get("DeviceIdentifier", "").lower() == disk_id.lower():
# Found our disk
for p in d.get("Partitions", []):
if p.get("Content", "").lower() == "efi":
return p.get("DeviceIdentifier", None)
return None
def mount_partition(self, disk, sudo=False):
disk_id = self.get_identifier(disk)
if not disk_id:
return None
if sudo:
# We need to create a new folder, then mount it manually
fst = self.get_disk_fs_type(disk_id)
if not fst:
# No detected fs
return None
vn = self.get_volume_name(disk_id)
if vn == "":
vn = "Untitled"
# Get safe volume name
if os.path.exists(os.path.join("/Volumes", vn)):
num = 1
while True:
if os.path.exists(os.path.join("/Volumes", vn + " " + str(num))):
num += 1
continue
break
vn = vn + " " + str(num)
# Create the dir, then mount
out = self.r.run([
{"args":["mkdir", os.path.join("/Volumes", vn)], "sudo":True, "show":True},
{"args":["mount", "-t", fst, "/dev/"+disk_id, os.path.join("/Volumes", vn)], "sudo":True, "show":True}
], True)
self._update_disks()
if len(out) and type(out[0]) is tuple:
out = out[-1] # Set out to the last output
if out[2] != 0:
# Non-zero exit code, clean up our mount point - if it exists
if os.path.exists(os.path.join("/Volumes", vn)):
self.r.run({"args":["rm", "-Rf", os.path.join("/Volumes", vn)], "sudo":True})
return out
out = self.r.run({"args":[self.diskutil, "mount", disk_id]})
self._update_disks()
return out
def unmount_partition(self, disk):
disk_id = self.get_identifier(disk)
if not disk_id:
return None
out = self.r.run({"args":[self.diskutil, "unmount", disk_id]})
self._update_disks()
return out
def is_mounted(self, disk):
disk_id = self.get_identifier(disk)
if not disk_id:
return None
m = self.get_mount_point(disk_id)
return (m != None and len(m))
def get_volumes(self):
# Returns a list object with all volumes from disks
return self.disks.get("VolumesFromDisks", [])
def _get_value_apfs(self, disk, field, default = None):
return self._get_value(disk, field, default, True)
def _get_value(self, disk, field, default = None, apfs_only = False):
disk_id = self.get_identifier(disk)
if not disk_id:
return None
# Takes a disk identifier, and returns the requested value
for d in self.disks.get("AllDisksAndPartitions", []):
for a in d.get("APFSVolumes", []):
if a.get("DeviceIdentifier", "").lower() == disk_id.lower():
return a.get(field, default)
if apfs_only:
# Skip looking at regular partitions
continue
if d.get("DeviceIdentifier", "").lower() == disk_id.lower():
return d.get(field, default)
for a in d.get("Partitions", []):
if a.get("DeviceIdentifier", "").lower() == disk_id.lower():
return a.get(field, default)
return None
# Getter methods
def get_content(self, disk):
return self._get_value(disk, "Content")
def get_volume_name(self, disk):
return self._get_value(disk, "VolumeName")
def get_volume_uuid(self, disk):
return self._get_value(disk, "VolumeUUID")
def get_disk_uuid(self, disk):
return self._get_value(disk, "DiskUUID")
def get_mount_point(self, disk):
return self._get_value(disk, "MountPoint")
def open_mount_point(self, disk, new_window = False):
disk_id = self.get_identifier(disk)
if not disk_id:
return None
mount = self.get_mount_point(disk_id)
if not mount:
return None
out = self.r.run({"args":["open", mount]})
return out[2] == 0
def get_mounted_volumes(self):
# Returns a list of mounted volumes
vol_list = self.r.run({"args":["ls", "-1", "/Volumes"]})[0].split("\n")
vol_list = [ x for x in vol_list if x != "" ]
return vol_list
def get_mounted_volume_dicts(self):
# Returns a list of dicts of name, identifier, mount point dicts
vol_list = []
for v in self.get_mounted_volumes():
i = self.get_identifier(os.path.join("/Volumes", v))
if i == None:
i = self.get_identifier("/")
if not self.get_volume_name(i) == v:
# Not valid and not our boot drive
continue
vol_list.append({
"name" : self.get_volume_name(i),
"identifier" : i,
"mount_point" : self.get_mount_point(i),
"disk_uuid" : self.get_disk_uuid(i),
"volume_uuid" : self.get_volume_uuid(i)
})
return vol_list
def get_disks_and_partitions_dict(self):
# Returns a list of dictionaries like so:
# { "disk0" : { "partitions" : [
# {
# "identifier" : "disk0s1",
# "name" : "EFI",
# "mount_point" : "/Volumes/EFI"
# }
# ] } }
disks = {}
for d in self.disks.get("AllDisks", []):
# Get the parent and make sure it has an entry
parent = self.get_parent(d)
top_disk = self.get_top_identifier(d)
if top_disk == d and not self.is_core_storage(d):
# Top level, skip
continue
# Not top level - make sure it's not an apfs container or core storage container
if self.is_apfs_container(d):
continue
if self.is_cs_container(d):
continue
if not parent in disks:
disks[parent] = { "partitions" : [] }
disks[parent]["partitions"].append({
"name" : self.get_volume_name(d),
"identifier" : d,
"mount_point" : self.get_mount_point(d),
"disk_uuid" : self.get_disk_uuid(d),
"volume_uuid" : self.get_volume_uuid(d)
})
return disks

96
Scripts/diskwin.py Normal file
View File

@ -0,0 +1,96 @@
import subprocess, plistlib, sys, os, time, json
sys.path.append(os.path.abspath(os.path.dirname(os.path.realpath(__file__))))
import run
class Disk:
def __init__(self):
self.r = run.Run()
self._update_disks()
def update(self):
self._update_disks()
def _update_disks(self):
self.disks = self.get_disks()
def get_disks(self):
# We hate windows... all of us.
#
# This has to be done in 3 commands,
# 1. To get the PHYSICALDISK entries, index, and model
# 2. To get the drive letter, volume name, fs, and size
# 3. To get some connection between them...
#
# May you all forgive me...
disks = self.r.run({"args":["wmic", "diskdrive", "get", "deviceid,model,index,size,partitions"]})[0]
disks = disks.replace("\r","").split("\n")[1:]
p_disks = {}
for d in disks:
ds = [x for x in d.split(" ") if len(x)]
if len(ds) < 5:
continue
p_disks[ds[1]] = {
"index":int(ds[1]),
"device":ds[0],
"model":" ".join(ds[2:-2]),
"size":int(ds[-1]),
"partitioncount":int(ds[-2]),
"type":0 # 0 = Unknown, 1 = No Root Dir, 2 = Removable, 3 = Local, 4 = Network, 5 = Disc, 6 = RAM disk
}
if not len(p_disks):
# Drat, nothing
return p_disks
# Let's find a shitty way to map this biz now
shit = self.r.run({"args":["wmic", "path", "Win32_LogicalDiskToPartition", "get", "antecedent,dependent"]})[0]
shit = shit.replace("\r","").split("\n")[1:]
for s in shit:
s = s.lower()
d = p = mp = None
try:
dp = s.split("deviceid=")[1].split('"')[1]
d = dp.split("disk #")[1].split(",")[0]
p = dp.split("partition #")[1]
mp = s.split("deviceid=")[2].split('"')[1].upper()
except:
pass
if any([d, p, mp]):
# Got *something*
if p_disks.get(d,None):
if not p_disks[d].get("partitions",None):
p_disks[d]["partitions"] = {}
p_disks[d]["partitions"][p] = {"letter":mp}
# Last attempt to do this - let's get the partition names!
parts = self.r.run({"args":["wmic", "logicaldisk", "get", "deviceid,filesystem,volumename,size,drivetype"]})[0]
parts = parts.replace("\r","").split("\n")[1:]
for p in parts:
ps = [x for x in p.split(" ") if len(x)]
if len(ps) < 2:
# Need the drive letter and disk type at minimum
continue
# Organize!
plt = ps[0] # get letter
ptp = ps[1] # get disk type
# Initialize
pfs = pnm = None
psz = -1 # Set to -1 initially for indeterminate size
try:
pfs = ps[2] # get file system
psz = ps[3] # get size
pnm = " ".join(ps[4:]) # get the rest in the name
except:
pass
for d in p_disks:
p_dict = p_disks[d]
for pr in p_dict.get("partitions",{}):
pr = p_dict["partitions"][pr]
if pr.get("letter","").upper() == plt.upper():
# Found it - set all attributes
pr["size"] = int(psz)
pr["file system"] = pfs
pr["name"] = pnm
# Also need to set the parent drive's type
p_dict["type"] = int(ptp)
break
return p_disks

156
Scripts/downloader.py
View File

@ -1,79 +1,79 @@
import sys, os, time
# Python-aware urllib stuff
if sys.version_info >= (3, 0):
from urllib.request import urlopen
else:
from urllib2 import urlopen
class Downloader:
def __init__(self):
return
def _progress_hook(self, response, bytes_so_far, total_size):
if total_size > 0:
percent = float(bytes_so_far) / total_size
percent = round(percent*100, 2)
sys.stdout.write("Downloaded {:,} of {:,} bytes ({:.2f}%)\r".format(bytes_so_far, total_size, percent))
else:
sys.stdout.write("Downloaded {:,} bytes\r".format(bytes_so_far))
def get_string(self, url, progress = True):
response = urlopen(url)
CHUNK = 16 * 1024
bytes_so_far = 0
try:
total_size = int(response.headers['Content-Length'])
except:
total_size = -1
chunk_so_far = "".encode("utf-8")
while True:
chunk = response.read(CHUNK)
bytes_so_far += len(chunk)
if progress:
self._progress_hook(response, bytes_so_far, total_size)
if not chunk:
break
chunk_so_far += chunk
return chunk_so_far.decode("utf-8")
def get_bytes(self, url, progress = True):
response = urlopen(url)
CHUNK = 16 * 1024
bytes_so_far = 0
try:
total_size = int(response.headers['Content-Length'])
except:
total_size = -1
chunk_so_far = "".encode("utf-8")
while True:
chunk = response.read(CHUNK)
bytes_so_far += len(chunk)
if progress:
self._progress_hook(response, bytes_so_far, total_size)
if not chunk:
break
chunk_so_far += chunk
return chunk_so_far
def stream_to_file(self, url, file, progress = True):
response = urlopen(url)
CHUNK = 16 * 1024
bytes_so_far = 0
try:
total_size = int(response.headers['Content-Length'])
except:
total_size = -1
with open(file, 'wb') as f:
while True:
chunk = response.read(CHUNK)
bytes_so_far += len(chunk)
if progress:
self._progress_hook(response, bytes_so_far, total_size)
if not chunk:
break
f.write(chunk)
if os.path.exists(file):
return file
else:
import sys, os, time
# Python-aware urllib stuff
if sys.version_info >= (3, 0):
from urllib.request import urlopen
else:
from urllib2 import urlopen
class Downloader:
def __init__(self):
return
def _progress_hook(self, response, bytes_so_far, total_size):
if total_size > 0:
percent = float(bytes_so_far) / total_size
percent = round(percent*100, 2)
sys.stdout.write("Downloaded {:,} of {:,} bytes ({:.2f}%)\r".format(bytes_so_far, total_size, percent))
else:
sys.stdout.write("Downloaded {:,} bytes\r".format(bytes_so_far))
def get_string(self, url, progress = True):
response = urlopen(url)
CHUNK = 16 * 1024
bytes_so_far = 0
try:
total_size = int(response.headers['Content-Length'])
except:
total_size = -1
chunk_so_far = "".encode("utf-8")
while True:
chunk = response.read(CHUNK)
bytes_so_far += len(chunk)
if progress:
self._progress_hook(response, bytes_so_far, total_size)
if not chunk:
break
chunk_so_far += chunk
return chunk_so_far.decode("utf-8")
def get_bytes(self, url, progress = True):
response = urlopen(url)
CHUNK = 16 * 1024
bytes_so_far = 0
try:
total_size = int(response.headers['Content-Length'])
except:
total_size = -1
chunk_so_far = "".encode("utf-8")
while True:
chunk = response.read(CHUNK)
bytes_so_far += len(chunk)
if progress:
self._progress_hook(response, bytes_so_far, total_size)
if not chunk:
break
chunk_so_far += chunk
return chunk_so_far
def stream_to_file(self, url, file, progress = True):
response = urlopen(url)
CHUNK = 16 * 1024
bytes_so_far = 0
try:
total_size = int(response.headers['Content-Length'])
except:
total_size = -1
with open(file, 'wb') as f:
while True:
chunk = response.read(CHUNK)
bytes_so_far += len(chunk)
if progress:
self._progress_hook(response, bytes_so_far, total_size)
if not chunk:
break
f.write(chunk)
if os.path.exists(file):
return file
else:
return None

364
Scripts/plist.py
View File

@ -1,182 +1,182 @@
### ###
# Imports #
### ###
import datetime
from io import BytesIO
import os
import plistlib
import struct
import sys
try:
FMT_XML = plistlib.FMT_XML
except:
FMT_XML = None
### ###
# Helper Methods #
### ###
def _check_py3():
return True if sys.version_info >= (3, 0) else False
def _is_binary(fp):
header = fp.read(32)
fp.seek(0)
return header[:8] == b'bplist00'
def _get_inst():
if _check_py3():
return (str)
else:
return (str, unicode)
### ###
# Deprecated Functions - Remapped #
### ###
def readPlist(pathOrFile):
if not isinstance(pathOrFile, _get_inst()):
return load(pathOrFile)
with open(pathOrFile, "rb") as f:
return load(f)
def writePlist(value, pathOrFile):
if not isinstance(pathOrFile, _get_inst()):
return dump(value, pathOrFile, fmt=FMT_XML, sort_keys=True, skipkeys=False)
with open(pathOrFile, "wb") as f:
return dump(value, f, fmt=FMT_XML, sort_keys=True, skipkeys=False)
### ###
# Remapped Functions #
### ###
def load(fp, fmt=None, use_builtin_types=True, dict_type=dict):
if _check_py3():
return plistlib.load(fp, fmt=fmt, use_builtin_types=use_builtin_types, dict_type=dict_type)
elif not _is_binary(fp):
return plistlib.readPlist(fp)
else:
return readBinaryPlistFile(fp)
def loads(value, fmt=None, use_builtin_types=True, dict_type=dict):
if _check_py3() and isinstance(value, _get_inst()):
# We were sent a string in py3 - let's encode it to some utf-8 bytes for fun!
value = value.encode()
fp = BytesIO(value)
if _check_py3():
return plistlib.load(fp, fmt=fmt, use_builtin_types=use_builtin_types, dict_type=dict_type)
elif not _is_binary(fp):
return plistlib.readPlistFromString(value)
else:
return readBinaryPlistFile(fp)
def dump(value, fp, fmt=FMT_XML, sort_keys=True, skipkeys=False):
if _check_py3():
plistlib.dump(value, fp, fmt=fmt, sort_keys=sort_keys, skipkeys=skipkeys)
else:
plistlib.writePlist(value, fp)
def dumps(value, fmt=FMT_XML, skipkeys=False):
if _check_py3():
return plistlib.dumps(value, fmt=fmt, skipkeys=skipkeys).encode("utf-8")
else:
return plistlib.writePlistToString(value).encode("utf-8")
### ###
# Binary Plist Stuff For Py2 #
### ###
# timestamp 0 of binary plists corresponds to 1/1/2001 (year of Mac OS X 10.0), instead of 1/1/1970.
MAC_OS_X_TIME_OFFSET = (31 * 365 + 8) * 86400
class InvalidFileException(ValueError):
def __str__(self):
return "Invalid file"
def __unicode__(self):
return "Invalid file"
def readBinaryPlistFile(in_file):
"""
Read a binary plist file, following the description of the binary format: http://opensource.apple.com/source/CF/CF-550/CFBinaryPList.c
Raise InvalidFileException in case of error, otherwise return the root object, as usual
Original patch diffed here: https://bugs.python.org/issue14455
"""
in_file.seek(-32, os.SEEK_END)
trailer = in_file.read(32)
if len(trailer) != 32:
return InvalidFileException()
offset_size, ref_size, num_objects, top_object, offset_table_offset = struct.unpack('>6xBB4xL4xL4xL', trailer)
in_file.seek(offset_table_offset)
object_offsets = []
offset_format = '>' + {1: 'B', 2: 'H', 4: 'L', 8: 'Q', }[offset_size] * num_objects
ref_format = {1: 'B', 2: 'H', 4: 'L', 8: 'Q', }[ref_size]
int_format = {0: (1, '>B'), 1: (2, '>H'), 2: (4, '>L'), 3: (8, '>Q'), }
object_offsets = struct.unpack(offset_format, in_file.read(offset_size * num_objects))
def getSize(token_l):
""" return the size of the next object."""
if token_l == 0xF:
m = ord(in_file.read(1)) & 0x3
s, f = int_format[m]
return struct.unpack(f, in_file.read(s))[0]
return token_l
def readNextObject(offset):
""" read the object at offset. May recursively read sub-objects (content of an array/dict/set) """
in_file.seek(offset)
token = in_file.read(1)
token_h, token_l = ord(token) & 0xF0, ord(token) & 0x0F #high and low parts
if token == '\x00':
return None
elif token == '\x08':
return False
elif token == '\x09':
return True
elif token == '\x0f':
return ''
elif token_h == 0x10: #int
result = 0
for k in xrange((2 << token_l) - 1):
result = (result << 8) + ord(in_file.read(1))
return result
elif token_h == 0x20: #real
if token_l == 2:
return struct.unpack('>f', in_file.read(4))[0]
elif token_l == 3:
return struct.unpack('>d', in_file.read(8))[0]
elif token_h == 0x30: #date
f = struct.unpack('>d', in_file.read(8))[0]
return datetime.datetime.utcfromtimestamp(f + MAC_OS_X_TIME_OFFSET)
elif token_h == 0x80: #data
s = getSize(token_l)
return in_file.read(s)
elif token_h == 0x50: #ascii string
s = getSize(token_l)
return in_file.read(s)
elif token_h == 0x60: #unicode string
s = getSize(token_l)
return in_file.read(s * 2).decode('utf-16be')
elif token_h == 0x80: #uid
return in_file.read(token_l + 1)
elif token_h == 0xA0: #array
s = getSize(token_l)
obj_refs = struct.unpack('>' + ref_format * s, in_file.read(s * ref_size))
return map(lambda x: readNextObject(object_offsets[x]), obj_refs)
elif token_h == 0xC0: #set
s = getSize(token_l)
obj_refs = struct.unpack('>' + ref_format * s, in_file.read(s * ref_size))
return set(map(lambda x: readNextObject(object_offsets[x]), obj_refs))
elif token_h == 0xD0: #dict
result = {}
s = getSize(token_l)
key_refs = struct.unpack('>' + ref_format * s, in_file.read(s * ref_size))
obj_refs = struct.unpack('>' + ref_format * s, in_file.read(s * ref_size))
for k, o in zip(key_refs, obj_refs):
key = readNextObject(object_offsets[k])
obj = readNextObject(object_offsets[o])
result[key] = obj
return result
raise InvalidFileException()
return readNextObject(object_offsets[top_object])
### ###
# Imports #
### ###
import datetime
from io import BytesIO
import os
import plistlib
import struct
import sys
try:
FMT_XML = plistlib.FMT_XML
except:
FMT_XML = None
### ###
# Helper Methods #
### ###
def _check_py3():
return True if sys.version_info >= (3, 0) else False
def _is_binary(fp):
header = fp.read(32)
fp.seek(0)
return header[:8] == b'bplist00'
def _get_inst():
if _check_py3():
return (str)
else:
return (str, unicode)
### ###
# Deprecated Functions - Remapped #
### ###
def readPlist(pathOrFile):
if not isinstance(pathOrFile, _get_inst()):
return load(pathOrFile)
with open(pathOrFile, "rb") as f:
return load(f)
def writePlist(value, pathOrFile):
if not isinstance(pathOrFile, _get_inst()):
return dump(value, pathOrFile, fmt=FMT_XML, sort_keys=True, skipkeys=False)
with open(pathOrFile, "wb") as f:
return dump(value, f, fmt=FMT_XML, sort_keys=True, skipkeys=False)
### ###
# Remapped Functions #
### ###
def load(fp, fmt=None, use_builtin_types=True, dict_type=dict):
if _check_py3():
return plistlib.load(fp, fmt=fmt, use_builtin_types=use_builtin_types, dict_type=dict_type)
elif not _is_binary(fp):
return plistlib.readPlist(fp)
else:
return readBinaryPlistFile(fp)
def loads(value, fmt=None, use_builtin_types=True, dict_type=dict):
if _check_py3() and isinstance(value, _get_inst()):
# We were sent a string in py3 - let's encode it to some utf-8 bytes for fun!
value = value.encode()
fp = BytesIO(value)
if _check_py3():
return plistlib.load(fp, fmt=fmt, use_builtin_types=use_builtin_types, dict_type=dict_type)
elif not _is_binary(fp):
return plistlib.readPlistFromString(value)
else:
return readBinaryPlistFile(fp)
def dump(value, fp, fmt=FMT_XML, sort_keys=True, skipkeys=False):
if _check_py3():
plistlib.dump(value, fp, fmt=fmt, sort_keys=sort_keys, skipkeys=skipkeys)
else:
plistlib.writePlist(value, fp)
def dumps(value, fmt=FMT_XML, skipkeys=False):
if _check_py3():
return plistlib.dumps(value, fmt=fmt, skipkeys=skipkeys).encode("utf-8")
else:
return plistlib.writePlistToString(value).encode("utf-8")
### ###
# Binary Plist Stuff For Py2 #
### ###
# timestamp 0 of binary plists corresponds to 1/1/2001 (year of Mac OS X 10.0), instead of 1/1/1970.
MAC_OS_X_TIME_OFFSET = (31 * 365 + 8) * 86400
class InvalidFileException(ValueError):
def __str__(self):
return "Invalid file"
def __unicode__(self):
return "Invalid file"
def readBinaryPlistFile(in_file):
"""
Read a binary plist file, following the description of the binary format: http://opensource.apple.com/source/CF/CF-550/CFBinaryPList.c
Raise InvalidFileException in case of error, otherwise return the root object, as usual
Original patch diffed here: https://bugs.python.org/issue14455
"""
in_file.seek(-32, os.SEEK_END)
trailer = in_file.read(32)
if len(trailer) != 32:
return InvalidFileException()
offset_size, ref_size, num_objects, top_object, offset_table_offset = struct.unpack('>6xBB4xL4xL4xL', trailer)
in_file.seek(offset_table_offset)
object_offsets = []
offset_format = '>' + {1: 'B', 2: 'H', 4: 'L', 8: 'Q', }[offset_size] * num_objects
ref_format = {1: 'B', 2: 'H', 4: 'L', 8: 'Q', }[ref_size]
int_format = {0: (1, '>B'), 1: (2, '>H'), 2: (4, '>L'), 3: (8, '>Q'), }
object_offsets = struct.unpack(offset_format, in_file.read(offset_size * num_objects))
def getSize(token_l):
""" return the size of the next object."""
if token_l == 0xF:
m = ord(in_file.read(1)) & 0x3
s, f = int_format[m]
return struct.unpack(f, in_file.read(s))[0]
return token_l
def readNextObject(offset):
""" read the object at offset. May recursively read sub-objects (content of an array/dict/set) """
in_file.seek(offset)
token = in_file.read(1)
token_h, token_l = ord(token) & 0xF0, ord(token) & 0x0F #high and low parts
if token == '\x00':
return None
elif token == '\x08':
return False
elif token == '\x09':
return True
elif token == '\x0f':
return ''
elif token_h == 0x10: #int
result = 0
for k in xrange((2 << token_l) - 1):
result = (result << 8) + ord(in_file.read(1))
return result
elif token_h == 0x20: #real
if token_l == 2:
return struct.unpack('>f', in_file.read(4))[0]
elif token_l == 3:
return struct.unpack('>d', in_file.read(8))[0]
elif token_h == 0x30: #date
f = struct.unpack('>d', in_file.read(8))[0]
return datetime.datetime.utcfromtimestamp(f + MAC_OS_X_TIME_OFFSET)
elif token_h == 0x80: #data
s = getSize(token_l)
return in_file.read(s)
elif token_h == 0x50: #ascii string
s = getSize(token_l)
return in_file.read(s)
elif token_h == 0x60: #unicode string
s = getSize(token_l)
return in_file.read(s * 2).decode('utf-16be')
elif token_h == 0x80: #uid
return in_file.read(token_l + 1)
elif token_h == 0xA0: #array
s = getSize(token_l)
obj_refs = struct.unpack('>' + ref_format * s, in_file.read(s * ref_size))
return map(lambda x: readNextObject(object_offsets[x]), obj_refs)
elif token_h == 0xC0: #set
s = getSize(token_l)
obj_refs = struct.unpack('>' + ref_format * s, in_file.read(s * ref_size))
return set(map(lambda x: readNextObject(object_offsets[x]), obj_refs))
elif token_h == 0xD0: #dict
result = {}
s = getSize(token_l)
key_refs = struct.unpack('>' + ref_format * s, in_file.read(s * ref_size))
obj_refs = struct.unpack('>' + ref_format * s, in_file.read(s * ref_size))
for k, o in zip(key_refs, obj_refs):
key = readNextObject(object_offsets[k])
obj = readNextObject(object_offsets[o])
result[key] = obj
return result
raise InvalidFileException()
return readNextObject(object_offsets[top_object])

150
Scripts/run.py Normal file
View File

@ -0,0 +1,150 @@
import sys
import subprocess
import threading
import shlex
try:
from Queue import Queue, Empty
except:
from queue import Queue, Empty
ON_POSIX = 'posix' in sys.builtin_module_names
class Run:
def __init__(self):
return
def _read_output(self, pipe, q):
while True:
try:
q.put(pipe.read(1))
except ValueError:
pipe.close()
break
def _stream_output(self, comm, shell = False):
output = error = ""
p = ot = et = None
try:
if shell and type(comm) is list:
comm = " ".join(shlex.quote(x) for x in comm)
if not shell and type(comm) is str:
comm = shlex.split(comm)
p = subprocess.Popen(comm, shell=shell, stdout=subprocess.PIPE, stderr=subprocess.PIPE, bufsize=1, universal_newlines=True, close_fds=ON_POSIX)
# Setup the stdout thread/queue
q = Queue()
t = threading.Thread(target=self._read_output, args=(p.stdout, q))
t.daemon = True # thread dies with the program
# Setup the stderr thread/queue
qe = Queue()
te = threading.Thread(target=self._read_output, args=(p.stderr, qe))
te.daemon = True # thread dies with the program
# Start both threads
t.start()
te.start()
while True:
c = z = ""
try:
c = q.get_nowait()
except Empty:
pass
else:
output += c
try:
z = qe.get_nowait()
except Empty:
pass
else:
error += z
sys.stdout.write(c)
sys.stdout.write(z)
sys.stdout.flush()
p.poll()
if c==z=="" and p.returncode != None:
break
o, e = p.communicate()
ot.exit()
et.exit()
return (output+o, error+e, p.returncode)
except:
if ot or et:
try: ot.exit()
except: pass
try: et.exit()
except: pass
if p:
return (output, error, p.returncode)
return ("", "Command not found!", 1)
def _run_command(self, comm, shell = False):
c = None
try:
if shell and type(comm) is list:
comm = " ".join(shlex.quote(x) for x in comm)
if not shell and type(comm) is str:
comm = shlex.split(comm)
p = subprocess.Popen(comm, shell=shell, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
c = p.communicate()
return (c[0].decode("utf-8", "ignore"), c[1].decode("utf-8", "ignore"), p.returncode)
except:
if c == None:
return ("", "Command not found!", 1)
return (c[0].decode("utf-8", "ignore"), c[1].decode("utf-8", "ignore"), p.returncode)
def run(self, command_list, leave_on_fail = False):
# Command list should be an array of dicts
if type(command_list) is dict:
# We only have one command
command_list = [command_list]
output_list = []
for comm in command_list:
args = comm.get("args", [])
shell = comm.get("shell", False)
stream = comm.get("stream", False)
sudo = comm.get("sudo", False)
stdout = comm.get("stdout", False)
stderr = comm.get("stderr", False)
mess = comm.get("message", None)
show = comm.get("show", False)
if not mess == None:
print(mess)
if not len(args):
# nothing to process
continue
if sudo:
# Check if we have sudo
out = self._run_command(["which", "sudo"])
if "sudo" in out[0]:
# Can sudo
if type(args) is list:
args.insert(0, out[0].replace("\n", "")) # add to start of list
elif type(args) is str:
args = out[0].replace("\n", "") + " " + args # add to start of string
if show:
print(" ".join(args))
if stream:
# Stream it!
out = self._stream_output(args, shell)
else:
# Just run and gather output
out = self._run_command(args, shell)
if stdout and len(out[0]):
print(out[0])
if stderr and len(out[1]):
print(out[1])
# Append output
output_list.append(out)
# Check for errors
if leave_on_fail and out[2] != 0:
# Got an error - leave
break
if len(output_list) == 1:
# We only ran one command - just return that output
return output_list[0]
return output_list

544
Scripts/utils.py
View File

@ -1,272 +1,272 @@
import sys, os, time, re, json, datetime, ctypes, subprocess
if os.name == "nt":
# Windows
import msvcrt
else:
# Not Windows \o/
import select
class Utils:
def __init__(self, name = "Python Script"):
self.name = name
# Init our colors before we need to print anything
cwd = os.getcwd()
os.chdir(os.path.dirname(os.path.realpath(__file__)))
if os.path.exists("colors.json"):
self.colors_dict = json.load(open("colors.json"))
else:
self.colors_dict = {}
os.chdir(cwd)
def check_admin(self):
# Returns whether or not we're admin
try:
is_admin = os.getuid() == 0
except AttributeError:
is_admin = ctypes.windll.shell32.IsUserAnAdmin() != 0
return is_admin
def elevate(self, file):
# Runs the passed file as admin
if self.check_admin():
return
if os.name == "nt":
ctypes.windll.shell32.ShellExecuteW(None, "runas", sys.executable, file, None, 1)
else:
try:
p = subprocess.Popen(["which", "sudo"], stdout=subprocess.PIPE, stderr=subprocess.PIPE)
c = p.communicate()[0].decode("utf-8", "ignore").replace("\n", "")
os.execv(c, [ sys.executable, 'python'] + sys.argv)
except:
exit(1)
def compare_versions(self, vers1, vers2, **kwargs):
# Helper method to compare ##.## strings
#
# vers1 < vers2 = True
# vers1 = vers2 = None
# vers1 > vers2 = False
# Sanitize the pads
pad = str(kwargs.get("pad", ""))
sep = str(kwargs.get("separator", "."))
ignore_case = kwargs.get("ignore_case", True)
# Cast as strings
vers1 = str(vers1)
vers2 = str(vers2)
if ignore_case:
vers1 = vers1.lower()
vers2 = vers2.lower()
# Split and pad lists
v1_parts, v2_parts = self.pad_length(vers1.split(sep), vers2.split(sep))
# Iterate and compare
for i in range(len(v1_parts)):
# Remove non-numeric
v1 = ''.join(c.lower() for c in v1_parts[i] if c.isalnum())
v2 = ''.join(c.lower() for c in v2_parts[i] if c.isalnum())
# Equalize the lengths
v1, v2 = self.pad_length(v1, v2)
# Compare
if str(v1) < str(v2):
return True
elif str(v1) > str(v2):
return False
# Never differed - return None, must be equal
return None
def pad_length(self, var1, var2, pad = "0"):
# Pads the vars on the left side to make them equal length
pad = "0" if len(str(pad)) < 1 else str(pad)[0]
if not type(var1) == type(var2):
# Type mismatch! Just return what we got
return (var1, var2)
if len(var1) < len(var2):
if type(var1) is list:
var1.extend([str(pad) for x in range(len(var2) - len(var1))])
else:
var1 = "{}{}".format((pad*(len(var2)-len(var1))), var1)
elif len(var2) < len(var1):
if type(var2) is list:
var2.extend([str(pad) for x in range(len(var1) - len(var2))])
else:
var2 = "{}{}".format((pad*(len(var1)-len(var2))), var2)
return (var1, var2)
def check_path(self, path):
# Loop until we either get a working path - or no changes
count = 0
while count < 100:
count += 1
if not len(path):
# We uh.. stripped out everything - bail
return None
if os.path.exists(path):
# Exists!
return os.path.abspath(path)
# Check quotes first
if (path[0] == '"' and path[-1] == '"') or (path[0] == "'" and path[-1] == "'"):
path = path[1:-1]
continue
# Check for tilde
if path[0] == "~":
test_path = os.path.expanduser(path)
if test_path != path:
# We got a change
path = test_path
continue
# If we have no spaces to trim - bail
if not (path[0] == " " or path[0] == " ") and not(path[-1] == " " or path[-1] == " "):
return None
# Here we try stripping spaces/tabs
test_path = path
t_count = 0
while t_count < 100:
t_count += 1
t_path = test_path
while len(t_path):
if os.path.exists(t_path):
return os.path.abspath(t_path)
if t_path[-1] == " " or t_path[-1] == " ":
t_path = t_path[:-1]
continue
break
if test_path[0] == " " or test_path[0] == " ":
test_path = test_path[1:]
continue
break
# Escapes!
test_path = "\\".join([x.replace("\\", "") for x in path.split("\\\\")])
if test_path != path and not (path[0] == " " or path[0] == " "):
path = test_path
continue
if path[0] == " " or path[0] == " ":
path = path[1:]
return None
def grab(self, prompt, **kwargs):
# Takes a prompt, a default, and a timeout and shows it with that timeout
# returning the result
timeout = kwargs.get("timeout", 0)
default = kwargs.get("default", None)
# If we don't have a timeout - then skip the timed sections
if timeout <= 0:
if sys.version_info >= (3, 0):
return input(prompt)
else:
return str(raw_input(prompt))
# Write our prompt
sys.stdout.write(prompt)
sys.stdout.flush()
if os.name == "nt":
start_time = time.time()
i = ''
while True:
if msvcrt.kbhit():
c = msvcrt.getche()
if ord(c) == 13: # enter_key
break
elif ord(c) >= 32: #space_char
i += c
if len(i) == 0 and (time.time() - start_time) > timeout:
break
else:
i, o, e = select.select( [sys.stdin], [], [], timeout )
if i:
i = sys.stdin.readline().strip()
print('') # needed to move to next line
if len(i) > 0:
return i
else:
return default
def cls(self):
os.system('cls' if os.name=='nt' else 'clear')
def cprint(self, message, **kwargs):
strip_colors = kwargs.get("strip_colors", False)
if os.name == "nt":
strip_colors = True
reset = u"\u001b[0m"
# Requires sys import
for c in self.colors:
if strip_colors:
message = message.replace(c["find"], "")
else:
message = message.replace(c["find"], c["replace"])
if strip_colors:
return message
sys.stdout.write(message)
print(reset)
# Needs work to resize the string if color chars exist
'''# Header drawing method
def head(self, text = None, width = 55):
if text == None:
text = self.name
self.cls()
print(" {}".format("#"*width))
len_text = self.cprint(text, strip_colors=True)
mid_len = int(round(width/2-len(len_text)/2)-2)
middle = " #{}{}{}#".format(" "*mid_len, len_text, " "*((width - mid_len - len(len_text))-2))
if len(middle) > width+1:
# Get the difference
di = len(middle) - width
# Add the padding for the ...#
di += 3
# Trim the string
middle = middle[:-di]
newlen = len(middle)
middle += "...#"
find_list = [ c["find"] for c in self.colors ]
# Translate colored string to len
middle = middle.replace(len_text, text + self.rt_color) # always reset just in case
self.cprint(middle)
print("#"*width)'''
# Header drawing method
def head(self, text = None, width = 55):
if text == None:
text = self.name
self.cls()
print(" {}".format("#"*width))
mid_len = int(round(width/2-len(text)/2)-2)
middle = " #{}{}{}#".format(" "*mid_len, text, " "*((width - mid_len - len(text))-2))
if len(middle) > width+1:
# Get the difference
di = len(middle) - width
# Add the padding for the ...#
di += 3
# Trim the string
middle = middle[:-di] + "...#"
print(middle)
print("#"*width)
def resize(self, width, height):
print('\033[8;{};{}t'.format(height, width))
def custom_quit(self):
self.head()
print("by CorpNewt\n")
print("Thanks for testing it out, for bugs/comments/complaints")
print("send me a message on Reddit, or check out my GitHub:\n")
print("www.reddit.com/u/corpnewt")
print("www.github.com/corpnewt\n")
# Get the time and wish them a good morning, afternoon, evening, and night
hr = datetime.datetime.now().time().hour
if hr > 3 and hr < 12:
print("Have a nice morning!\n\n")
elif hr >= 12 and hr < 17:
print("Have a nice afternoon!\n\n")
elif hr >= 17 and hr < 21:
print("Have a nice evening!\n\n")
else:
print("Have a nice night!\n\n")
exit(0)
import sys, os, time, re, json, datetime, ctypes, subprocess
if os.name == "nt":
# Windows
import msvcrt
else:
# Not Windows \o/
import select
class Utils:
def __init__(self, name = "Python Script"):
self.name = name
# Init our colors before we need to print anything
cwd = os.getcwd()
os.chdir(os.path.dirname(os.path.realpath(__file__)))
if os.path.exists("colors.json"):
self.colors_dict = json.load(open("colors.json"))
else:
self.colors_dict = {}
os.chdir(cwd)
def check_admin(self):
# Returns whether or not we're admin
try:
is_admin = os.getuid() == 0
except AttributeError:
is_admin = ctypes.windll.shell32.IsUserAnAdmin() != 0
return is_admin
def elevate(self, file):
# Runs the passed file as admin
if self.check_admin():
return
if os.name == "nt":
ctypes.windll.shell32.ShellExecuteW(None, "runas", sys.executable, file, None, 1)
else:
try:
p = subprocess.Popen(["which", "sudo"], stdout=subprocess.PIPE, stderr=subprocess.PIPE)
c = p.communicate()[0].decode("utf-8", "ignore").replace("\n", "")
os.execv(c, [ sys.executable, 'python'] + sys.argv)
except:
exit(1)
def compare_versions(self, vers1, vers2, **kwargs):
# Helper method to compare ##.## strings
#
# vers1 < vers2 = True
# vers1 = vers2 = None
# vers1 > vers2 = False
# Sanitize the pads
pad = str(kwargs.get("pad", ""))
sep = str(kwargs.get("separator", "."))
ignore_case = kwargs.get("ignore_case", True)
# Cast as strings
vers1 = str(vers1)
vers2 = str(vers2)
if ignore_case:
vers1 = vers1.lower()
vers2 = vers2.lower()
# Split and pad lists
v1_parts, v2_parts = self.pad_length(vers1.split(sep), vers2.split(sep))
# Iterate and compare
for i in range(len(v1_parts)):
# Remove non-numeric
v1 = ''.join(c.lower() for c in v1_parts[i] if c.isalnum())
v2 = ''.join(c.lower() for c in v2_parts[i] if c.isalnum())
# Equalize the lengths
v1, v2 = self.pad_length(v1, v2)
# Compare
if str(v1) < str(v2):
return True
elif str(v1) > str(v2):
return False
# Never differed - return None, must be equal
return None
def pad_length(self, var1, var2, pad = "0"):
# Pads the vars on the left side to make them equal length
pad = "0" if len(str(pad)) < 1 else str(pad)[0]
if not type(var1) == type(var2):
# Type mismatch! Just return what we got
return (var1, var2)
if len(var1) < len(var2):
if type(var1) is list:
var1.extend([str(pad) for x in range(len(var2) - len(var1))])
else:
var1 = "{}{}".format((pad*(len(var2)-len(var1))), var1)
elif len(var2) < len(var1):
if type(var2) is list:
var2.extend([str(pad) for x in range(len(var1) - len(var2))])
else:
var2 = "{}{}".format((pad*(len(var1)-len(var2))), var2)
return (var1, var2)
def check_path(self, path):
# Loop until we either get a working path - or no changes
count = 0
while count < 100:
count += 1
if not len(path):
# We uh.. stripped out everything - bail
return None
if os.path.exists(path):
# Exists!
return os.path.abspath(path)
# Check quotes first
if (path[0] == '"' and path[-1] == '"') or (path[0] == "'" and path[-1] == "'"):
path = path[1:-1]
continue
# Check for tilde
if path[0] == "~":
test_path = os.path.expanduser(path)
if test_path != path:
# We got a change
path = test_path
continue
# If we have no spaces to trim - bail
if not (path[0] == " " or path[0] == " ") and not(path[-1] == " " or path[-1] == " "):
return None
# Here we try stripping spaces/tabs
test_path = path
t_count = 0
while t_count < 100:
t_count += 1
t_path = test_path
while len(t_path):
if os.path.exists(t_path):
return os.path.abspath(t_path)
if t_path[-1] == " " or t_path[-1] == " ":
t_path = t_path[:-1]
continue
break
if test_path[0] == " " or test_path[0] == " ":
test_path = test_path[1:]
continue
break
# Escapes!
test_path = "\\".join([x.replace("\\", "") for x in path.split("\\\\")])
if test_path != path and not (path[0] == " " or path[0] == " "):
path = test_path
continue
if path[0] == " " or path[0] == " ":
path = path[1:]
return None
def grab(self, prompt, **kwargs):
# Takes a prompt, a default, and a timeout and shows it with that timeout
# returning the result
timeout = kwargs.get("timeout", 0)
default = kwargs.get("default", None)
# If we don't have a timeout - then skip the timed sections
if timeout <= 0:
if sys.version_info >= (3, 0):
return input(prompt)
else:
return str(raw_input(prompt))
# Write our prompt
sys.stdout.write(prompt)
sys.stdout.flush()
if os.name == "nt":
start_time = time.time()
i = ''
while True:
if msvcrt.kbhit():
c = msvcrt.getche()
if ord(c) == 13: # enter_key
break
elif ord(c) >= 32: #space_char
i += c
if len(i) == 0 and (time.time() - start_time) > timeout:
break
else:
i, o, e = select.select( [sys.stdin], [], [], timeout )
if i:
i = sys.stdin.readline().strip()
print('') # needed to move to next line
if len(i) > 0:
return i
else:
return default
def cls(self):
os.system('cls' if os.name=='nt' else 'clear')
def cprint(self, message, **kwargs):
strip_colors = kwargs.get("strip_colors", False)
if os.name == "nt":
strip_colors = True
reset = u"\u001b[0m"
# Requires sys import
for c in self.colors:
if strip_colors:
message = message.replace(c["find"], "")
else:
message = message.replace(c["find"], c["replace"])
if strip_colors:
return message
sys.stdout.write(message)
print(reset)
# Needs work to resize the string if color chars exist
'''# Header drawing method
def head(self, text = None, width = 55):
if text == None:
text = self.name
self.cls()
print(" {}".format("#"*width))
len_text = self.cprint(text, strip_colors=True)
mid_len = int(round(width/2-len(len_text)/2)-2)
middle = " #{}{}{}#".format(" "*mid_len, len_text, " "*((width - mid_len - len(len_text))-2))
if len(middle) > width+1:
# Get the difference
di = len(middle) - width
# Add the padding for the ...#
di += 3
# Trim the string
middle = middle[:-di]
newlen = len(middle)
middle += "...#"
find_list = [ c["find"] for c in self.colors ]
# Translate colored string to len
middle = middle.replace(len_text, text + self.rt_color) # always reset just in case
self.cprint(middle)
print("#"*width)'''
# Header drawing method
def head(self, text = None, width = 55):
if text == None:
text = self.name
self.cls()
print(" {}".format("#"*width))
mid_len = int(round(width/2-len(text)/2)-2)
middle = " #{}{}{}#".format(" "*mid_len, text, " "*((width - mid_len - len(text))-2))
if len(middle) > width+1:
# Get the difference
di = len(middle) - width
# Add the padding for the ...#
di += 3
# Trim the string
middle = middle[:-di] + "...#"
print(middle)
print("#"*width)
def resize(self, width, height):
print('\033[8;{};{}t'.format(height, width))
def custom_quit(self):
self.head()
print("by CorpNewt\n")
print("Thanks for testing it out, for bugs/comments/complaints")
print("send me a message on Reddit, or check out my GitHub:\n")
print("www.reddit.com/u/corpnewt")
print("www.github.com/corpnewt\n")
# Get the time and wish them a good morning, afternoon, evening, and night
hr = datetime.datetime.now().time().hour
if hr > 3 and hr < 12:
print("Have a nice morning!\n\n")
elif hr >= 12 and hr < 17:
print("Have a nice afternoon!\n\n")
elif hr >= 17 and hr < 21:
print("Have a nice evening!\n\n")
else:
print("Have a nice night!\n\n")
exit(0)