commoncode/src/commoncode/fileutils.py at main · aboutcode-org/commoncode · GitHub

History

578 lines (461 loc) · 16.8 KB

Raw

100

101

102

103

104

105

106

107

108

109

110

111

112

113

114

115

116

117

118

119

120

121

122

123

124

125

126

127

128

129

130

131

132

133

134

135

136

137

138

139

140

141

142

143

144

145

146

147

148

149

150

151

152

153

154

155

156

157

158

159

160

161

162

163

164

165

166

167

168

169

170

171

172

173

174

175

176

177

178

179

180

181

182

183

184

185

186

187

188

189

190

191

192

193

194

195

196

197

198

199

200

201

202

203

204

205

206

207

208

209

210

211

212

213

214

215

216

217

218

219

220

221

222

223

224

225

226

227

228

229

230

231

232

233

234

235

236

237

238

239

240

241

242

243

244

245

246

247

248

249

250

251

252

253

254

255

256

257

258

259

260

261

262

263

264

265

266

267

268

269

270

271

272

273

274

275

276

277

278

279

280

281

282

283

284

285

286

287

288

289

290

291

292

293

294

295

296

297

298

299

300

301

302

303

304

305

306

307

308

309

310

311

312

313

314

315

316

317

318

319

320

321

322

323

324

325

326

327

328

329

330

331

332

333

334

335

336

337

338

339

340

341

342

343

344

345

346

347

348

349

350

351

352

353

354

355

356

357

358

359

360

361

362

363

364

365

366

367

368

369

370

371

372

373

374

375

376

377

378

379

380

381

382

383

384

385

386

387

388

389

390

391

392

393

394

395

396

397

398

399

400

401

402

403

404

405

406

407

408

409

410

411

412

413

414

415

416

417

418

419

420

421

422

423

424

425

426

427

428

429

430

431

432

433

434

435

436

437

438

439

440

441

442

443

444

445

446

447

448

449

450

451

452

453

454

455

456

457

458

459

460

461

462

463

464

465

466

467

468

469

470

471

472

473

474

475

476

477

478

479

480

481

482

483

484

485

486

487

488

489

490

491

492

493

494

495

496

497

498

499

500

501

502

503

504

505

506

507

508

509

510

511

512

513

514

515

516

517

518

519

520

521

522

523

524

525

526

527

528

529

530

531

532

533

534

535

536

537

538

539

540

541

542

543

544

545

546

547

548

549

550

551

552

553

554

555

556

557

558

559

560

561

562

563

564

565

566

567

568

569

570

571

572

573

574

575

576

577

578

# SPDX-License-Identifier: Apache-2.0 AND Python-2.0

# See http://www.apache.org/licenses/LICENSE-2.0 for the license text.

# See https://github.com/aboutcode-org/commoncode for support or download.

# See https://aboutcode.org for more information about nexB OSS projects.

import errno

import ntpath

import os

import posixpath

import shutil

import stat

import sys

import tempfile

from os import fsdecode

try:

from scancode_config import scancode_temp_dir as _base_temp_dir

except ImportError:

_base_temp_dir = None

from commoncode import filetype

from commoncode.filetype import is_rwx

# this exception is not available on posix

try:

WindowsError # NOQA

except NameError:

class WindowsError(Exception):

pass

import logging

logger = logging.getLogger(__name__)

TRACE = False

def logger_debug(*args):

pass

if TRACE:

logging.basicConfig(stream=sys.stdout)

logger.setLevel(logging.DEBUG)

def logger_debug(*args):

return logger.debug(" ".join(isinstance(a, str) and a or repr(a) for a in args))

"""

File, paths and directory utility functions.

"""

# DIRECTORIES

def create_dir(location):

"""

Create directory and all sub-directories recursively at location ensuring these

are readable and writeable.

Raise Exceptions if it fails to create the directory.

"""

if os.path.exists(location):

if not os.path.isdir(location):

err = "Cannot create directory: existing file in the way %(location)s."

raise OSError(err % locals())

else:

# may fail on win if the path is too long

# FIXME: consider using UNC ?\\ paths

try:

os.makedirs(location)

chmod(location, RW, recurse=False)

# avoid multi-process TOCTOU conditions when creating dirs

# the directory may have been created since the exist check

except WindowsError as e:

# [Error 183] Cannot create a file when that file already exists

if e and e.winerror == 183:

if not os.path.isdir(location):

raise

else:

raise

except (IOError, OSError) as o:

if o.errno == errno.EEXIST:

if not os.path.isdir(location):

raise

else:

raise

def get_temp_dir(base_dir=_base_temp_dir, prefix=""):

"""

Return the path to a new existing unique temporary directory, created under

the `base_dir` base directory using the `prefix` prefix.

If `base_dir` is not provided, use the 'SCANCODE_TMP' env var or the system

temp directory.

WARNING: do not change this code without changing scancode_config.py too

"""

has_base = bool(base_dir)

if not has_base:

base_dir = os.getenv("SCANCODE_TMP")

if not base_dir:

base_dir = tempfile.gettempdir()

if not os.path.exists(base_dir):

create_dir(base_dir)

if not has_base:

prefix = "scancode-tk-"

return tempfile.mkdtemp(prefix=prefix, dir=base_dir)

# PATHS AND NAMES MANIPULATIONS

# TODO: move these functions to paths.py

def prepare_path(pth):

"""

Return the `pth` path string either as encoded bytes if on Linux and using

Python 2 or as a unicode/text otherwise.

"""

if not isinstance(pth, str):

return fsdecode(pth)

return pth

def is_posixpath(location):

r"""

Return True if the `location` path is likely a POSIX-like path using POSIX path

separators (slash or "/")or has no path separator.

Return False if the `location` path is likely a Windows-like path using backslash

as path separators (e.g. "\").

"""

has_slashes = "/" in location

has_backslashes = "\\" in location

# windows paths with drive

if location:

drive, _ = ntpath.splitdrive(location)

if drive:

return False

# a path is always POSIX unless it contains ONLY backslahes

# which is a rough approximation (it could still be posix)

is_posix = True

if has_backslashes and not has_slashes:

is_posix = False

return is_posix

def as_posixpath(location):

r"""

Return a POSIX-like path using POSIX path separators (slash or "/") for a

`location` path. This converts Windows paths to look like POSIX paths: Python

accepts gracefully POSIX paths on Windows.

"""

location = prepare_path(location)

return location.replace("\\", "/")

def as_winpath(location):

r"""

Return a Windows-like path using Windows path separators (backslash or "\") for a

`location` path.

"""

location = prepare_path(location)

return location.replace("/", "\\")

def split_parent_resource(path, force_posix=False):

"""

Return a tuple of (parent directory path, resource name).

"""

use_posix = force_posix or is_posixpath(path)

splitter = use_posix and posixpath or ntpath

path_no_trailing_speps = path.rstrip("\\/")

return splitter.split(path_no_trailing_speps)

def resource_name(path, force_posix=False):

"""

Return the resource name (file name or directory name) from `path` which

is the last path segment.

"""

_left, right = split_parent_resource(path, force_posix)

return right or ""

def file_name(path, force_posix=False):

"""

Return the file name (or directory name) of a path.

"""

return resource_name(path, force_posix)

def parent_directory(path, force_posix=False, with_trail=True):

"""

Return the parent directory path of a file or directory `path`.

The returned directory has a trailing path separator unless with_trail is False.

"""

left, _right = split_parent_resource(path, force_posix)

use_posix = force_posix or is_posixpath(path)

sep = "/" if use_posix else "\\"

trail = sep if with_trail and left != sep else ""

return left + trail

def file_base_name(path, force_posix=False):

"""

Return the file base name for a path. The base name is the base name of

the file minus the extension. For a directory return an empty string.

"""

return splitext(path, force_posix)[0]

def file_extension(path, force_posix=False):

"""

Return the file extension for a path.

"""

return splitext(path, force_posix)[1]

def splitext_name(file_name, is_file=True):

"""

Return a tuple of Unicode strings (basename, extension) for a file name. The

basename is the file name minus its extension. Return an empty extension

string for a directory. Not the same as os.path.splitext_name.

"""

if not file_name:

return "", ""

file_name = fsdecode(file_name)

if not is_file:

return file_name, ""

if file_name.startswith(".") and "." not in file_name[1:]:

# .dot files base name is the full name and they do not have an extension

return file_name, ""

base_name, extension = posixpath.splitext(file_name)

# handle composed extensions of tar.gz, bz, zx,etc

if base_name.endswith(".tar"):

base_name, extension2 = posixpath.splitext(base_name)

extension = extension2 + extension

return base_name, extension

# TODO: FIXME: this is badly broken!!!!

def splitext(path, force_posix=False):

"""

Return a tuple of strings (basename, extension) for a path. The basename is

the file name minus its extension. Return an empty extension string for a

directory.

"""

base_name = ""

extension = ""

if not path:

return base_name, extension

is_dir = path.endswith(

(

"\\",

"/",

)

path = as_posixpath(path).strip("/")

name = resource_name(path, force_posix)

if is_dir:

# directories never have an extension

base_name = name

extension = ""

elif name.startswith(".") and "." not in name[1:]:

# .dot files base name is the full name and they do not have an extension

base_name = name

extension = ""

else:

base_name, extension = posixpath.splitext(name)

# handle composed extensions of tar.gz, tar.bz2, zx,etc

if base_name.endswith(".tar"):

base_name, extension2 = posixpath.splitext(base_name)

extension = extension2 + extension

return base_name, extension

# DIRECTORY AND FILES WALKING/ITERATION

def ignore_nothing(_):

return False

def walk(location, ignored=None, follow_symlinks=False):

"""

Walk location returning the same tuples as os.walk but with a different

behavior:

- always walk top-down, breadth-first.

- ignore and do not follow symlinks unless `follow_symlinks` is True,

- always ignore special files (FIFOs, etc.)

- optionally ignore files and directories by invoking the `ignored`

callable on files and directories returning True if it should be ignored.

- location is a directory or a file: for a file, the file is returned.

If `follow_symlinks` is True, then symlinks will not be ignored and be

collected like regular files and directories

"""

is_ignored = ignored(location) if ignored else False

if is_ignored:

if TRACE:

logger_debug("walk: ignored:", location, is_ignored)

return

if filetype.is_file(location, follow_symlinks=follow_symlinks):

yield parent_directory(location), [], [file_name(location)]

elif filetype.is_dir(location, follow_symlinks=follow_symlinks):

dirs = []

files = []

for resource in os.scandir(location):

loc = os.path.join(location, resource.name)

if filetype.is_special(loc) or (ignored and ignored(loc)):

if (

follow_symlinks

and resource.is_symlink()

and not filetype.is_broken_link(location)

pass

else:

if TRACE:

ign = ignored and ignored(loc)

logger_debug("walk: ignored:", loc, ign)

continue

# special files and symlinks are always ignored

if resource.is_dir(follow_symlinks=follow_symlinks):

dirs.append(resource.name)

elif resource.is_file(follow_symlinks=follow_symlinks):

files.append(resource.name)

yield location, dirs, files

for dr in dirs:

for tripple in walk(

os.path.join(location, dr), ignored, follow_symlinks=follow_symlinks

yield tripple

def resource_iter(location, ignored=ignore_nothing, with_dirs=True, follow_symlinks=False):

"""

Return an iterable of paths at `location` recursively.

:param location: a file or a directory.

:param ignored: a callable accepting a location argument and returning True

if the location should be ignored.

:return: an iterable of file and directory locations.

"""

for top, dirs, files in walk(location, ignored, follow_symlinks=follow_symlinks):

if with_dirs:

for d in dirs:

yield os.path.join(top, d)

for f in files:

yield os.path.join(top, f)

# COPY

def copytree(src, dst):

"""

Copy recursively the `src` directory to the `dst` directory. If `dst` is an

existing directory, files in `dst` may be overwritten during the copy.

Preserve timestamps.

Ignores:

-`src` permissions: `dst` files are created with the default permissions.

- all special files such as FIFO or character devices and symlinks.

Raise an shutil.Error with a list of reasons.

This function is similar to and derived from the Python shutil.copytree

function. See fileutils.py.ABOUT for details.

"""

if not filetype.is_readable(src):

chmod(src, R, recurse=False)

names = [resource.name for resource in os.scandir(src)]

if not os.path.exists(dst):

os.makedirs(dst)

errors = []

errors.extend(copytime(src, dst))

for name in names:

srcname = os.path.join(src, name)

dstname = os.path.join(dst, name)

# skip anything that is not a regular file, dir or link

if not filetype.is_regular(srcname):

continue

if not filetype.is_readable(srcname):

chmod(srcname, R, recurse=False)

try:

if os.path.isdir(srcname):

copytree(srcname, dstname)

elif filetype.is_file(srcname):

copyfile(srcname, dstname)

# catch the Error from the recursive copytree so that we can

# continue with other files

except shutil.Error as err:

errors.extend(err.args[0])

except EnvironmentError as why:

errors.append((srcname, dstname, str(why)))

if errors:

raise shutil.Error(errors)

def copyfile(src, dst):

"""

Copy src file to dst file preserving timestamps.

Ignore permissions and special files.

Similar to and derived from Python shutil module. See fileutils.py.ABOUT

for details.

"""

if not filetype.is_regular(src):

return

if not filetype.is_readable(src):

chmod(src, R, recurse=False)

if os.path.isdir(dst):

dst = os.path.join(dst, os.path.basename(src))

shutil.copyfile(src, dst)

copytime(src, dst)

def copytime(src, dst):

"""

Copy timestamps from `src` to `dst`.

Similar to and derived from Python shutil module. See fileutils.py.ABOUT

for details.

"""

errors = []

st = os.stat(src)

if hasattr(os, "utime"):

try:

os.utime(dst, (st.st_atime, st.st_mtime))

except OSError as why:

if WindowsError is not None and isinstance(why, WindowsError):

# File access times cannot be copied on Windows

pass

else:

errors.append((src, dst, str(why)))

return errors

# PERMISSIONS

# modes: read, write, executable

R = stat.S_IRUSR

RW = stat.S_IRUSR | stat.S_IWUSR

RX = stat.S_IRUSR | stat.S_IXUSR

RWX = stat.S_IRUSR | stat.S_IWUSR | stat.S_IXUSR

# FIXME: This was an expensive operation that used to recurse of the parent directory

def chmod(location, flags, recurse=False):

"""

Update permissions for `location` with with `flags`. `flags` is one of R,

RW, RX or RWX with the same semantics as in the chmod command. Update is

done recursively if `recurse`.

"""

if not location or not os.path.exists(location):

return

location = os.path.abspath(location)

new_flags = flags

if filetype.is_dir(location):

# POSIX dirs need to be executable to be readable,

# and to be writable so we can change perms of files inside

new_flags = RWX

# FIXME: do we really need to change the parent directory perms?

# FIXME: may just check them instead?

parent = os.path.dirname(location)

current_stat = stat.S_IMODE(os.stat(parent).st_mode)

if not is_rwx(parent):

os.chmod(parent, current_stat | RWX)

if filetype.is_regular(location):

current_stat = stat.S_IMODE(os.stat(location).st_mode)

os.chmod(location, current_stat | new_flags)

if recurse:

chmod_tree(location, flags)

def chmod_tree(location, flags):

"""

Update permissions recursively in a directory tree `location`.

"""

if filetype.is_dir(location):

for top, dirs, files in walk(location):

for d in dirs:

chmod(os.path.join(top, d), flags, recurse=False)

for f in files:

chmod(os.path.join(top, f), flags, recurse=False)

# DELETION

def _rm_handler(function, path, excinfo): # NOQA

"""

shutil.rmtree handler invoked on error when deleting a directory tree.

This retries deleting once before giving up.

"""

if TRACE:

logger_debug("_rm_handler:", "path:", path, "excinfo:", excinfo)

if function in (os.rmdir, os.listdir, os.scandir):

try:

chmod(path, RW, recurse=True)

shutil.rmtree(path, True)

except Exception:

pass

if os.path.exists(path):

logger.warning("Failed to delete directory %s", path)

elif function == os.remove:

try:

delete(path, _err_handler=None)

except Exception:

pass

if os.path.exists(path):

logger.warning("Failed to delete file %s", path)

def delete(location, _err_handler=_rm_handler):

"""

Delete a directory or file at `location` recursively. Similar to "rm -rf"

in a shell or a combo of os.remove and shutil.rmtree.

This function is design to never fails: instead it logs warnings if the

file or directory was not deleted correctly.

"""

if not location:

return

if os.path.exists(location) or filetype.is_broken_link(location):

if filetype.is_dir(location):

shutil.rmtree(location, False, _rm_handler)

else:

os.remove(location)

commoncode/src/commoncode/fileutils.py at main · aboutcode-org/commoncode · GitHub

Uh oh!

Uh oh!

commoncode/src/commoncode/fileutils.py at main · aboutcode-org/commoncode · GitHub

Uh oh!

Uh oh!

Expand file tree

Breadcrumbs

fileutils.py

Latest commit

History

Breadcrumbs

fileutils.py

File metadata and controls